[draft] skeleton: components module from dynamic text input

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -8,6 +8,7 @@
 import importlib
 import inspect
 import re
+import types
 from functools import partial
 from typing import (
     Any,
@@ -986,8 +987,11 @@ def create_custom_component(self, model: Any, config: Config, **kwargs: Any) ->
         :param config: The custom defined connector config
         :return: The declarative component built from the Pydantic model to be used at runtime
         """
-
-        custom_component_class = self._get_class_from_fully_qualified_class_name(model.class_name)
+        components_module = self._get_components_module_object(config=config)
+        custom_component_class = self._get_class_from_fully_qualified_class_name(
+            full_qualified_class_name=model.class_name,
+            components_module=components_module,
+        )
         component_fields = get_type_hints(custom_component_class)
         model_args = model.dict()
         model_args["config"] = config
@@ -1039,15 +1043,59 @@ def create_custom_component(self, model: Any, config: Config, **kwargs: Any) ->
         }
         return custom_component_class(**kwargs)
 
-    @staticmethod
-    def _get_class_from_fully_qualified_class_name(full_qualified_class_name: str) -> Any:
+    def _get_components_module_object(
+        config: Config,
+    ) -> None:
+        """Get a components module object based on the provided config.
+
+        If custom python components is provided, this will be loaded. Otherwise, we will
+        attempt to load from the `components` module already imported.
+        """
+        INJECTED_COMPONENTS_PY = "__injected_components_py"
+        COMPONENTS_MODULE_NAME = "components"
+
+        components_module: types.ModuleType
+        if INJECTED_COMPONENTS_PY in config:
+            # Create a new module object and execute the provided Python code text within it
+            components_module = types.ModuleType(name=COMPONENTS_MODULE_NAME)
+            python_text = config[INJECTED_COMPONENTS_PY]
+            exec(python_text, components_module.__dict__)
+            # Skip insert the module into sys.modules because we pass by reference below
+            # sys.modules[module_name] = components_module
+        else:
+            components_module = importlib.import_module(name=COMPONENTS_MODULE_NAME)
+
+    def _get_class_from_fully_qualified_class_name(
+        full_qualified_class_name: str,
+        components_module: types.ModuleType,
+    ) -> Any:
+        """
+        Get a class from its fully qualified name, optionally using a pre-parsed module.
+
+        Args:
+            full_qualified_class_name (str): The fully qualified name of the class (e.g., "module.ClassName").
+            components_module (Optional[ModuleType]): An optional pre-parsed module.
+
+        Returns:
+            Any: The class object.
+
+        Raises:
+            ValueError: If the class cannot be loaded.
+        """
         split = full_qualified_class_name.split(".")
-        module = ".".join(split[:-1])
+        module_name_full = ".".join(split[:-1])
+        module_name = split[:-2]
         class_name = split[-1]
+
+        if module_name != "components":
+            raise ValueError(
+                f"Custom components must be defined in a module named `components`. Found {module_name} instead."
+            )
+
         try:
-            return getattr(importlib.import_module(module), class_name)
-        except AttributeError:
-            raise ValueError(f"Could not load class {full_qualified_class_name}.")
+            return getattr(components_module, class_name)
+        except (AttributeError, ModuleNotFoundError) as e:
+            raise ValueError(f"Could not load class {full_qualified_class_name}.") from e
 
     @staticmethod
     def _derive_component_type_from_type_hints(field_type: Any) -> Optional[str]:

airbyte_cdk/test/utils/manifest_only_fixtures.py

@@ -2,6 +2,7 @@
 
 
 import importlib.util
+import types
 from pathlib import Path
 from types import ModuleType
 from typing import Optional
@@ -51,6 +52,25 @@ def components_module(connector_dir: Path) -> Optional[ModuleType]:
     return components_module
 
 
+def components_module_from_string(components_py_text: str) -> Optional[ModuleType]:
+    """Load and return the components module from a provided string containing the python code.
+
+    This assumes the components module is located at <connector_dir>/components.py.
+
+    TODO: Make new unit test to leverage this fixture
+    """
+    module_name = "components"
+
+    # Create a new module object
+    components_module = types.ModuleType(name=module_name)
+
+    # Execute the module text in the module's namespace
+    exec(components_py_text, components_module.__dict__)
+
+    # Now you can import and use the module
+    return components_module
+
+
 @pytest.fixture(scope="session")
 def manifest_path(connector_dir: Path) -> Path:
     """Return the path to the connector's manifest file."""

unit_tests/source_declarative_manifest/conftest.py

@@ -2,13 +2,26 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 #
 
+import hashlib
 import os
+from pathlib import Path
+from typing import Any, Literal
 
 import pytest
 import yaml
 
 
-def get_fixture_path(file_name):
+def hash_text(input_text: str, hash_type: Literal["md5", "sha256"] = "md5") -> str:
+    hashers = {
+        "md5": hashlib.md5,
+        "sha256": hashlib.sha256,
+    }
+    hash_object = hashers[hash_type]()
+    hash_object.update(input_text.encode())
+    return hash_object.hexdigest()
+
+
+def get_fixture_path(file_name) -> str:
     return os.path.join(os.path.dirname(__file__), file_name)
 
 
@@ -52,3 +65,21 @@ def valid_local_config_file():
 @pytest.fixture
 def invalid_local_config_file():
     return get_fixture_path("resources/invalid_local_pokeapi_config.json")
+
+
+@pytest.fixture
+def py_components_config_dict() -> dict[str, Any]:
+    manifest_dict = yaml.safe_load(
+        get_fixture_path("resources/valid_py_components.yaml"),
+    )
+    custom_py_code_path = get_fixture_path("resources/valid_py_components_code.py")
+    custom_py_code = Path(custom_py_code_path).read_text()
+    combined_config_dict = {
+        "__injected_declarative_manifest": manifest_dict,
+        "__injected_components_py": custom_py_code,
+        "__injected_components_py_checksum": {
+            "md5": hash_text(custom_py_code, "md5"),
+            "sha256": hash_text(custom_py_code, "sha256"),
+        },
+    }
+    return combined_config_dict

unit_tests/source_declarative_manifest/resources/valid_py_components_code.py

@@ -0,0 +1,15 @@
+"""Custom Python components.py file for testing.
+
+This file is mostly a no-op (for now) but should trigger a failure if code file is not
+correctly parsed.
+"""
+
+from airbyte_cdk.sources.declarative.models import DeclarativeStream
+
+
+class CustomDeclarativeStream(DeclarativeStream):
+    """Custom declarative stream class.
+
+    We don't change anything from the base class, but this should still be enough to confirm
+    that the components.py file is correctly parsed.
+    """

unit_tests/source_declarative_manifest/resources/valid_py_components_config.json

@@ -0,0 +1,3 @@
+{
+  "pokemon_name": "blastoise"
+}