Instantiate the transform/check classes. (#198)

* Instantiate the transform/check classes. Temporary fix.

* checks: Do not include branch or client when instantiating to follow .

* Remove unused transform/check loading functions. Change branch to empty string when instantiating transform class.

* Fix linting/formatting after conflict fix in GitHub UI.

Author: FragmentedPacket

infrahub_sdk/checks.py

@@ -11,13 +11,10 @@
 from git.repo import Repo
 from pydantic import BaseModel, Field
 
-from .exceptions import InfrahubCheckNotFoundError, UninitializedError
+from .exceptions import UninitializedError
 
 if TYPE_CHECKING:
-    from pathlib import Path
-
     from . import InfrahubClient
-    from .schema.repository import InfrahubCheckDefinitionConfig
 
 INFRAHUB_CHECK_VARIABLE_TO_IMPORT = "INFRAHUB_CHECKS"
 
@@ -176,27 +173,3 @@ async def run(self, data: dict | None = None) -> bool:
             self.log_info("Check succesfully completed")
 
         return self.passed
-
-
-def get_check_class_instance(
-    check_config: InfrahubCheckDefinitionConfig, search_path: Path | None = None
-) -> InfrahubCheck:
-    if check_config.file_path.is_absolute() or search_path is None:
-        search_location = check_config.file_path
-    else:
-        search_location = search_path / check_config.file_path
-
-    try:
-        spec = importlib.util.spec_from_file_location(check_config.class_name, search_location)
-        module = importlib.util.module_from_spec(spec)  # type: ignore[arg-type]
-        spec.loader.exec_module(module)  # type: ignore[union-attr]
-
-        # Get the specified class from the module
-        check_class = getattr(module, check_config.class_name)
-
-        # Create an instance of the class
-        check_instance = check_class()
-    except (FileNotFoundError, AttributeError) as exc:
-        raise InfrahubCheckNotFoundError(name=check_config.name) from exc
-
-    return check_instance

infrahub_sdk/pytest_plugin/items/check.py

@@ -37,9 +37,10 @@ def instantiate_check(self) -> None:
             str(self.resource_config.file_path.parent) if self.resource_config.file_path.parent != Path() else None  # type: ignore[attr-defined]
         )
 
-        self.check_instance = self.resource_config.load_class(  # type: ignore[attr-defined]
+        check_class = self.resource_config.load_class(  # type: ignore[attr-defined]
             import_root=self.repository_base, relative_path=relative_path
         )
+        self.check_instance = check_class()
 
     def run_check(self, variables: dict[str, Any]) -> Any:
         self.instantiate_check()

infrahub_sdk/pytest_plugin/items/python_transform.py

@@ -36,9 +36,12 @@ def instantiate_transform(self) -> None:
         relative_path = (
             str(self.resource_config.file_path.parent) if self.resource_config.file_path.parent != Path() else None  # type: ignore[attr-defined]
         )
-        self.transform_instance = self.resource_config.load_class(  # type: ignore[attr-defined]
+        transform_class = self.resource_config.load_class(  # type: ignore[attr-defined]
             import_root=self.repository_base, relative_path=relative_path
         )
+        client = self.session.infrahub_client  # type: ignore[attr-defined]
+        # TODO: Look into seeing how a transform class may use the branch, but set as a empty string for the time being to keep current behaviour
+        self.transform_instance = transform_class(branch="", client=client)
 
     def run_transform(self, variables: dict[str, Any]) -> Any:
         self.instantiate_transform()

infrahub_sdk/transforms.py

@@ -1,20 +1,16 @@
 from __future__ import annotations
 
 import asyncio
-import importlib
 import os
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Any
 
 from git import Repo
 
-from .exceptions import InfrahubTransformNotFoundError, UninitializedError
+from .exceptions import UninitializedError
 
 if TYPE_CHECKING:
-    from pathlib import Path
-
     from . import InfrahubClient
-    from .schema.repository import InfrahubPythonTransformConfig
 
 INFRAHUB_TRANSFORM_VARIABLE_TO_IMPORT = "INFRAHUB_TRANSFORMS"
 
@@ -95,40 +91,3 @@ async def run(self, data: dict | None = None) -> Any:
             return await self.transform(data=unpacked)
 
         return self.transform(data=unpacked)
-
-
-def get_transform_class_instance(
-    transform_config: InfrahubPythonTransformConfig,
-    search_path: Path | None = None,
-    branch: str = "",
-    client: InfrahubClient | None = None,
-) -> InfrahubTransform:
-    """Gets an instance of the InfrahubTransform class.
-
-    Args:
-        transform_config: A config object with information required to find and load the transform.
-        search_path: The path in which to search for a python file containing the transform. The current directory is
-            assumed if not speicifed.
-        branch: Infrahub branch which will be targeted in graphql query used to acquire data for transformation.
-        client: InfrahubClient used to interact with infrahub API.
-    """
-    if transform_config.file_path.is_absolute() or search_path is None:
-        search_location = transform_config.file_path
-    else:
-        search_location = search_path / transform_config.file_path
-
-    try:
-        spec = importlib.util.spec_from_file_location(transform_config.class_name, search_location)
-        module = importlib.util.module_from_spec(spec)  # type: ignore[arg-type]
-        spec.loader.exec_module(module)  # type: ignore[union-attr]
-
-        # Get the specified class from the module
-        transform_class = getattr(module, transform_config.class_name)
-
-        # Create an instance of the class
-        transform_instance = transform_class(branch=branch, client=client)
-
-    except (FileNotFoundError, AttributeError) as exc:
-        raise InfrahubTransformNotFoundError(name=transform_config.name) from exc
-
-    return transform_instance