Add infrahub.tasks as built-in logger to InfrahubGenerator class (#…

…274)

* Test building in logger to InfrahubGenerator class to allow logs to be easily propagated via CTL and CI pipeline logs.

* Remove unused logging import.

* Add changelog to describe change.

changelog/+add-logger-to-generator-class.md

@@ -0,0 +1 @@
+Added logger to `InfrahubGenerator` class to allow users use built-in logging (`self.logger`) to show logging within Infrahub CI pipeline.

infrahub_sdk/ctl/generator.py

@@ -9,7 +9,7 @@
 from ..ctl import config
 from ..ctl.client import initialize_client
 from ..ctl.repository import get_repository_config
-from ..ctl.utils import execute_graphql_query, parse_cli_vars
+from ..ctl.utils import execute_graphql_query, init_logging, parse_cli_vars
 from ..exceptions import ModuleImportError
 from ..node import InfrahubNode
 
@@ -20,11 +20,12 @@
 async def run(
     generator_name: str,
     path: str,  # noqa: ARG001
-    debug: bool,  # noqa: ARG001
+    debug: bool,
     list_available: bool,
     branch: str | None = None,
     variables: list[str] | None = None,
 ) -> None:
+    init_logging(debug=debug)
     repository_config = get_repository_config(Path(config.INFRAHUB_REPO_CONFIG_FILE))
 
     if list_available or not generator_name:
@@ -34,7 +35,6 @@ async def run(
     generator_config = repository_config.get_generator_definition(name=generator_name)
 
     console = Console()
-
     relative_path = str(generator_config.file_path.parent) if generator_config.file_path.parent != Path() else None
 
     try:

infrahub_sdk/generator.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import os
 from abc import abstractmethod
 from typing import TYPE_CHECKING
@@ -27,6 +28,7 @@ def __init__(
         generator_instance: str = "",
         params: dict | None = None,
         convert_query_response: bool = False,
+        logger: logging.Logger | None = None,
     ) -> None:
         self.query = query
         self.branch = branch
@@ -41,6 +43,7 @@ def __init__(
         self._related_nodes: list[InfrahubNode] = []
         self.infrahub_node = infrahub_node
         self.convert_query_response = convert_query_response
+        self.logger = logger if logger else logging.getLogger("infrahub.tasks")
 
     @property
     def store(self) -> NodeStore: