feat(core): allow to dynamically inject external (private or public) plugins extending Python SDK dynamically

Author: helmut-hoffer-von-ankershoffen

src/aignostics/utils/__init__.py

@@ -22,7 +22,7 @@
     __version__,
     __version_full__,
 )
-from ._di import load_modules, locate_implementations, locate_subclasses
+from ._di import discover_plugin_packages, load_modules, locate_implementations, locate_subclasses
 from ._fs import get_user_data_directory, open_user_data_directory, sanitize_path, sanitize_path_component
 from ._health import Health
 from ._log import LogSettings
@@ -60,6 +60,7 @@
     "__version_full__",
     "boot",
     "console",
+    "discover_plugin_packages",
     "get_process_info",
     "get_user_data_directory",
     "load_modules",

src/aignostics/utils/_di.py

@@ -2,6 +2,8 @@
 
 import importlib
 import pkgutil
+from functools import lru_cache
+from importlib.metadata import entry_points
 from inspect import isclass
 from typing import Any
 
@@ -10,6 +12,28 @@
 _implementation_cache: dict[Any, list[Any]] = {}
 _subclass_cache: dict[Any, list[Any]] = {}
 
+# Entry point group name for aignostics plugins
+PLUGIN_ENTRY_POINT_GROUP = "aignostics.plugins"
+
+
+@lru_cache(maxsize=1)
+def discover_plugin_packages() -> tuple[str, ...]:
+    """
+    Discover plugin packages using entry points.
+
+    Plugins register themselves in their pyproject.toml:
+
+        [project.entry-points."aignostics.plugins"]
+        my_plugin = "my_plugin"
+
+    Results are cached after the first call.
+
+    Returns:
+        tuple[str, ...]: Tuple of discovered plugin package names.
+    """
+    eps = entry_points(group=PLUGIN_ENTRY_POINT_GROUP)
+    return tuple(ep.value for ep in eps)
+
 
 def load_modules() -> None:
     package = importlib.import_module(__project_name__)
@@ -21,6 +45,8 @@ def locate_implementations(_class: type[Any]) -> list[Any]:
     """
     Dynamically discover all instances of some class.
 
+    Searches in the main project and all plugins registered via entry points.
+
     Args:
         _class (type[Any]): Class to search for.
 
@@ -30,16 +56,19 @@ def locate_implementations(_class: type[Any]) -> list[Any]:
     if _class in _implementation_cache:
         return _implementation_cache[_class]
 
+    plugin_packages = discover_plugin_packages()
+
     implementations = []
-    package = importlib.import_module(__project_name__)
+    for package_name in [*plugin_packages, __project_name__]:
+        package = importlib.import_module(package_name)
 
-    for _, name, _ in pkgutil.iter_modules(package.__path__):
-        module = importlib.import_module(f"{__project_name__}.{name}")
-        # Check all members of the module
-        for member_name in dir(module):
-            member = getattr(module, member_name)
-            if isinstance(member, _class):
-                implementations.append(member)
+        for _, name, _ in pkgutil.iter_modules(package.__path__):
+            module = importlib.import_module(f"{package_name}.{name}")
+            # Check all members of the module
+            for member_name in dir(module):
+                member = getattr(module, member_name)
+                if isinstance(member, _class):
+                    implementations.append(member)
 
     _implementation_cache[_class] = implementations
     return implementations
@@ -49,6 +78,8 @@ def locate_subclasses(_class: type[Any]) -> list[Any]:
     """
     Dynamically discover all classes that are subclasses of some type.
 
+    Searches in the main project and all plugins registered via entry points.
+
     Args:
         _class (type[Any]): Parent class of subclasses to search for.
 
@@ -58,19 +89,25 @@ def locate_subclasses(_class: type[Any]) -> list[Any]:
     if _class in _subclass_cache:
         return _subclass_cache[_class]
 
-    subclasses = []
-    package = importlib.import_module(__project_name__)
+    plugin_packages = discover_plugin_packages()
 
-    for _, name, _ in pkgutil.iter_modules(package.__path__):
+    subclasses = []
+    for package_name in [*plugin_packages, __project_name__]:
         try:
-            module = importlib.import_module(f"{__project_name__}.{name}")
-            # Check all members of the module
-            for member_name in dir(module):
-                member = getattr(module, member_name)
-                if isclass(member) and issubclass(member, _class) and member != _class:
-                    subclasses.append(member)
+            package = importlib.import_module(package_name)
         except ImportError:
             continue
 
+        for _, name, _ in pkgutil.iter_modules(package.__path__):
+            try:
+                module = importlib.import_module(f"{package_name}.{name}")
+                # Check all members of the module
+                for member_name in dir(module):
+                    member = getattr(module, member_name)
+                    if isclass(member) and issubclass(member, _class) and member != _class:
+                        subclasses.append(member)
+            except ImportError:
+                continue
+
     _subclass_cache[_class] = subclasses
     return subclasses