Fix relative module imports

Fixes #81

Author: ogenstad

infrahub_sdk/ctl/cli_commands.py

@@ -36,14 +36,13 @@
     parse_cli_vars,
 )
 from ..ctl.validate import app as validate_app
-from ..exceptions import GraphQLError, InfrahubTransformNotFoundError
+from ..exceptions import GraphQLError, ModuleImportError
 from ..jinja2 import identify_faulty_jinja_code
 from ..schema import (
     InfrahubRepositoryConfig,
     MainSchemaTypes,
     SchemaRoot,
 )
-from ..transforms import get_transform_class_instance
 from ..utils import get_branch, write_to_file
 from ..yaml import SchemaFile
 from .exporter import dump
@@ -338,16 +337,17 @@ def transform(
     client = initialize_client()
 
     # Get python transform class instance
+
+    relative_path = str(transform_config.file_path.parent) if transform_config.file_path.parent != Path() else None
+
     try:
-        transform = get_transform_class_instance(
-            transform_config=transform_config,
-            branch=branch,
-            client=client,
-        )
-    except InfrahubTransformNotFoundError as exc:
-        console.print(f"Unable to load {transform_name} from python_transforms")
+        transform_class = transform_config.load_class(import_root=str(Path.cwd()), relative_path=relative_path)
+    except ModuleImportError as exc:
+        console.print(f"[red]{exc.message}")
         raise typer.Exit(1) from exc
 
+    transform = transform_class(client=client, branch=branch)
+
     # Get data
     query_str = repository_config.get_query(name=transform.query).load_query()
     data = asyncio.run(

infrahub_sdk/schema.py

@@ -14,6 +14,7 @@
 from .exceptions import InvalidResponseError, ModuleImportError, SchemaNotFoundError, ValidationError
 from .generator import InfrahubGenerator
 from .graphql import Mutation
+from .transforms import InfrahubTransform
 from .utils import duplicates
 
 if TYPE_CHECKING:
@@ -120,6 +121,21 @@ class InfrahubPythonTransformConfig(InfrahubRepositoryConfigElement):
     file_path: Path = Field(..., description="The file within the repository with the transform code.")
     class_name: str = Field(default="Transform", description="The name of the transform class to run.")
 
+    def load_class(
+        self, import_root: Optional[str] = None, relative_path: Optional[str] = None
+    ) -> type[InfrahubTransform]:
+        module = import_module(module_path=self.file_path, import_root=import_root, relative_path=relative_path)
+
+        if self.class_name not in dir(module):
+            raise ModuleImportError(message=f"The specified class {self.class_name} was not found within the module")
+
+        transform_class = getattr(module, self.class_name)
+
+        if not issubclass(transform_class, InfrahubTransform):
+            raise ModuleImportError(message=f"The specified class {self.class_name} is not an Infrahub Transform")
+
+        return transform_class
+
 
 class InfrahubRepositoryGraphQLConfig(InfrahubRepositoryConfigElement):
     model_config = ConfigDict(extra="forbid")

infrahub_sdk/transforms.py

@@ -3,18 +3,17 @@
 import asyncio
 import importlib
 import os
-import warnings
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Any, Optional
 
 from git import Repo
 
-from . import InfrahubClient
-from .exceptions import InfrahubTransformNotFoundError
+from .exceptions import InfrahubTransformNotFoundError, UninitializedError
 
 if TYPE_CHECKING:
     from pathlib import Path
 
+    from . import InfrahubClient
     from .schema import InfrahubPythonTransformConfig
 
 INFRAHUB_TRANSFORM_VARIABLE_TO_IMPORT = "INFRAHUB_TRANSFORMS"
@@ -48,25 +47,10 @@ def __init__(
 
     @property
     def client(self) -> InfrahubClient:
-        if not self._client:
-            self._client = InfrahubClient(address=self.server_url)
+        if self._client:
+            return self._client
 
-        return self._client
-
-    @classmethod
-    async def init(cls, client: Optional[InfrahubClient] = None, *args: Any, **kwargs: Any) -> InfrahubTransform:
-        """Async init method, If an existing InfrahubClient client hasn't been provided, one will be created automatically."""
-        warnings.warn(
-            f"{cls.__class__.__name__}.init has been deprecated and will be removed in the version after Infrahub SDK 1.0.0",
-            DeprecationWarning,
-            stacklevel=1,
-        )
-        if client:
-            kwargs["client"] = client
-
-        item = cls(*args, **kwargs)
-
-        return item
+        raise UninitializedError("The client has not been initialized")
 
     @property
     def branch_name(self) -> str: