feat: implement dependency injection for lazy loading of submodules and enhance module resolution

Author: endomorphosis

ipfs_kit_py/__init__.py

@@ -979,8 +979,28 @@ def get_p2p_workflow_tools():
 
 # Submodule lazy imports
 # These are optional dependencies available as submodules in the root directory
-def get_ipfs_datasets():
-    """Lazy import of ipfs_datasets_py submodule."""
+def get_ipfs_datasets(*, deps: object | None = None, module_override=None):
+    """Lazy import of ipfs_datasets_py submodule.
+
+    Supports dependency injection via ``module_override`` or a deps container
+    that implements ``get_cached``/``set_cached`` (e.g., ipfs_datasets_py.RouterDeps).
+    """
+    if module_override is not None:
+        setter = getattr(deps, "set_cached", None)
+        if callable(setter):
+            try:
+                setter("ipfs_kit_py::ipfs_datasets_py", module_override)
+            except Exception:
+                pass
+        return module_override
+    getter = getattr(deps, "get_cached", None)
+    if callable(getter):
+        try:
+            cached = getter("ipfs_kit_py::ipfs_datasets_py")
+            if cached is not None:
+                return cached
+        except Exception:
+            pass
     try:
         import sys
         from pathlib import Path
@@ -991,14 +1011,40 @@ def get_ipfs_datasets():
             sys.path.insert(0, str(datasets_path))
 
         import ipfs_datasets_py
+        setter = getattr(deps, "set_cached", None)
+        if callable(setter):
+            try:
+                setter("ipfs_kit_py::ipfs_datasets_py", ipfs_datasets_py)
+            except Exception:
+                pass
         return ipfs_datasets_py
     except ImportError as e:
         logger.debug(f"ipfs_datasets_py submodule not available: {e}")
         return None
 
 
-def get_ipfs_accelerate():
-    """Lazy import of ipfs_accelerate_py submodule."""
+def get_ipfs_accelerate(*, deps: object | None = None, module_override=None):
+    """Lazy import of ipfs_accelerate_py submodule.
+
+    Supports dependency injection via ``module_override`` or a deps container
+    that implements ``get_cached``/``set_cached`` (e.g., ipfs_datasets_py.RouterDeps).
+    """
+    if module_override is not None:
+        setter = getattr(deps, "set_cached", None)
+        if callable(setter):
+            try:
+                setter("ipfs_kit_py::ipfs_accelerate_py", module_override)
+            except Exception:
+                pass
+        return module_override
+    getter = getattr(deps, "get_cached", None)
+    if callable(getter):
+        try:
+            cached = getter("ipfs_kit_py::ipfs_accelerate_py")
+            if cached is not None:
+                return cached
+        except Exception:
+            pass
     try:
         import sys
         from pathlib import Path
@@ -1009,6 +1055,12 @@ def get_ipfs_accelerate():
             sys.path.insert(0, str(accelerate_path))
 
         import ipfs_accelerate_py
+        setter = getattr(deps, "set_cached", None)
+        if callable(setter):
+            try:
+                setter("ipfs_kit_py::ipfs_accelerate_py", ipfs_accelerate_py)
+            except Exception:
+                pass
         return ipfs_accelerate_py
     except ImportError as e:
         logger.debug(f"ipfs_accelerate_py submodule not available: {e}")

ipfs_kit_py/arrow_metadata_index.py

@@ -64,21 +64,34 @@
     get_ipfs_datasets_manager = None
     logger.info("ipfs_datasets_py not available - dataset storage disabled")
 
-# Import ipfs_accelerate_py for compute acceleration
-try:
-    import sys
-    from pathlib import Path as PathlibPath
-    accelerate_path = PathlibPath(__file__).parent.parent / "ipfs_accelerate_py"
-    if accelerate_path.exists():
-        sys.path.insert(0, str(accelerate_path))
-    
-    from ipfs_accelerate_py import AccelerateCompute
-    HAS_ACCELERATE = True
-    logger.info("ipfs_accelerate_py compute layer available for arrow metadata")
-except ImportError:
-    HAS_ACCELERATE = False
-    AccelerateCompute = None
-    logger.info("ipfs_accelerate_py not available - using default compute")
+HAS_ACCELERATE = False
+AccelerateCompute = None
+
+
+def _load_accelerate_compute_class(*, deps: object | None = None):
+    """Best-effort lazy loader for AccelerateCompute.
+
+    Supports dependency injection by allowing callers to pass a deps container
+    that may already cache an imported ipfs_accelerate_py module.
+    """
+
+    global HAS_ACCELERATE, AccelerateCompute
+    if AccelerateCompute is not None:
+        return AccelerateCompute
+    try:
+        from ipfs_kit_py import get_ipfs_accelerate
+
+        mod = get_ipfs_accelerate(deps=deps)
+        if mod is None:
+            return None
+        cls = getattr(mod, "AccelerateCompute", None)
+        if cls is None:
+            return None
+        AccelerateCompute = cls
+        HAS_ACCELERATE = True
+        return AccelerateCompute
+    except Exception:
+        return None
 
 #
 logger = logging.getLogger(__name__)
@@ -110,7 +123,10 @@ def __init__(
         cluster_id: str = "default",
         enable_dataset_storage: bool = False,
         enable_compute_layer: bool = False,
-        dataset_batch_size: int = 100
+        dataset_batch_size: int = 100,
+        *,
+        deps: object | None = None,
+        accelerate_compute=None,
     ):  # Added node_id and cluster_id
         # Check if PyArrow is available before proceeding
         if not ARROW_AVAILABLE:
@@ -152,8 +168,8 @@ def __init__(
         self.dataset_manager = None
         self._metadata_buffer = []
 
-        # Compute layer configuration  
-        self.enable_compute_layer = enable_compute_layer and HAS_ACCELERATE
+        # Compute layer configuration
+        self.enable_compute_layer = bool(enable_compute_layer)
         self.compute_layer = None
 
         # Initialize dataset manager if enabled
@@ -168,7 +184,13 @@ def __init__(
         # Initialize compute layer if enabled
         if self.enable_compute_layer:
             try:
-                self.compute_layer = AccelerateCompute()
+                if accelerate_compute is not None:
+                    self.compute_layer = accelerate_compute
+                else:
+                    cls = _load_accelerate_compute_class(deps=deps)
+                    if cls is None:
+                        raise ImportError("ipfs_accelerate_py not available")
+                    self.compute_layer = cls()
                 logger.info("Arrow Metadata Index compute layer enabled")
             except Exception as e:
                 logger.warning(f"Failed to initialize compute layer: {e}")

ipfs_kit_py/deps_resolver.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import importlib
+from typing import Any
+
+
+def deps_get(deps: object | None, key: str) -> Any | None:
+    if deps is None or not key:
+        return None
+    getter = getattr(deps, "get_cached", None)
+    if callable(getter):
+        try:
+            return getter(key)
+        except Exception:
+            return None
+    if isinstance(deps, dict):
+        return deps.get(key)
+    return None
+
+
+def deps_set(deps: object | None, key: str, value: Any) -> Any:
+    if deps is None or not key:
+        return value
+    setter = getattr(deps, "set_cached", None)
+    if callable(setter):
+        try:
+            return setter(key, value)
+        except Exception:
+            return value
+    if isinstance(deps, dict):
+        deps[key] = value
+    return value
+
+
+def resolve_module(
+    module_name: str,
+    *,
+    deps: object | None = None,
+    module_override: Any | None = None,
+    cache_key: str | None = None,
+) -> Any | None:
+    """Resolve a Python module with optional injection + deps caching."""
+    if module_override is not None:
+        deps_set(deps, cache_key or f"pip::{module_name}", module_override)
+        return module_override
+
+    cached = deps_get(deps, cache_key or f"pip::{module_name}")
+    if cached is not None:
+        return cached
+
+    try:
+        module = importlib.import_module(module_name)
+    except Exception:
+        return None
+
+    deps_set(deps, cache_key or f"pip::{module_name}", module)
+    return module
+
+
+def once(deps: object | None, key: str) -> bool:
+    """Return True the first time per deps container, False thereafter."""
+    if deps is None or not key:
+        return True
+    if deps_get(deps, key):
+        return False
+    deps_set(deps, key, True)
+    return True

ipfs_kit_py/enhanced_bucket_index.py

@@ -37,20 +37,28 @@
     get_ipfs_datasets_manager = None
     logger.info("ipfs_datasets_py not available - dataset storage disabled")
 
-# Import ipfs_accelerate_py for compute acceleration
-try:
-    import sys
-    accelerate_path = Path(__file__).parent.parent / "ipfs_accelerate_py"
-    if accelerate_path.exists():
-        sys.path.insert(0, str(accelerate_path))
-    
-    from ipfs_accelerate_py import AccelerateCompute
-    HAS_ACCELERATE = True
-    logger.info("ipfs_accelerate_py compute layer available")
-except ImportError:
-    HAS_ACCELERATE = False
-    AccelerateCompute = None
-    logger.info("ipfs_accelerate_py not available - using default compute")
+HAS_ACCELERATE = False
+AccelerateCompute = None
+
+
+def _load_accelerate_compute_class(*, deps: object | None = None):
+    global HAS_ACCELERATE, AccelerateCompute
+    if AccelerateCompute is not None:
+        return AccelerateCompute
+    try:
+        from ipfs_kit_py import get_ipfs_accelerate
+
+        mod = get_ipfs_accelerate(deps=deps)
+        if mod is None:
+            return None
+        cls = getattr(mod, "AccelerateCompute", None)
+        if cls is None:
+            return None
+        AccelerateCompute = cls
+        HAS_ACCELERATE = True
+        return AccelerateCompute
+    except Exception:
+        return None
 
 @dataclass
 class BucketMetadata:
@@ -93,7 +101,10 @@ def __init__(
         bucket_vfs_manager=None,
         enable_dataset_storage: bool = False,
         enable_compute_layer: bool = False,
-        dataset_batch_size: int = 100
+        dataset_batch_size: int = 100,
+        *,
+        deps: object | None = None,
+        accelerate_compute=None,
     ):
         """
         Initialize the enhanced bucket index.
@@ -123,7 +134,7 @@ def __init__(
         self._index_buffer = []
 
         # Compute layer configuration
-        self.enable_compute_layer = enable_compute_layer and HAS_ACCELERATE
+        self.enable_compute_layer = bool(enable_compute_layer)
         self.compute_layer = None
 
         # Initialize dataset manager if enabled
@@ -138,7 +149,13 @@ def __init__(
         # Initialize compute layer if enabled
         if self.enable_compute_layer:
             try:
-                self.compute_layer = AccelerateCompute()
+                if accelerate_compute is not None:
+                    self.compute_layer = accelerate_compute
+                else:
+                    cls = _load_accelerate_compute_class(deps=deps)
+                    if cls is None:
+                        raise ImportError("ipfs_accelerate_py not available")
+                    self.compute_layer = cls()
                 logger.info("Enhanced Bucket Index compute layer enabled")
             except Exception as e:
                 logger.warning(f"Failed to initialize compute layer: {e}")

ipfs_kit_py/ipfs_kit.py

@@ -372,6 +372,10 @@ def __init__(
         enable_cluster_management=False,
         enable_metadata_index=False,
         auto_start_daemons=True,
+        *,
+        deps: object | None = None,
+        ipfs_accelerate=None,
+        ipfs_datasets=None,
     ):
         """
         Initializes the IPFS Kit instance.
@@ -408,6 +412,24 @@ def __init__(
         # passing metadata.
         metadata = self.metadata
 
+        # Dependency injection container and optional injected cross-package modules.
+        self.deps = deps or (resources.get("deps") if isinstance(resources, dict) else None) or self.metadata.get("deps")
+        self.ipfs_accelerate_py = ipfs_accelerate or (
+            resources.get("ipfs_accelerate_py") if isinstance(resources, dict) else None
+        )
+        self.ipfs_datasets_py = ipfs_datasets or (resources.get("ipfs_datasets_py") if isinstance(resources, dict) else None)
+
+        # Best-effort: cache injected modules on deps for other packages to reuse.
+        setter = getattr(self.deps, "set_cached", None)
+        if callable(setter):
+            try:
+                if self.ipfs_accelerate_py is not None:
+                    setter("ipfs_kit_py::ipfs_accelerate_py", self.ipfs_accelerate_py)
+                if self.ipfs_datasets_py is not None:
+                    setter("ipfs_kit_py::ipfs_datasets_py", self.ipfs_datasets_py)
+            except Exception:
+                pass
+
         # Initialize _initialized attribute
         self._initialized = False
 
@@ -529,9 +551,8 @@ def __init__(
                     self.huggingface_kit = huggingface_kit(resources=resources, metadata=metadata)
                 else:
                     self.huggingface_kit = None # Initialize to None
-                    
+
                 # Initialize ipget component
-                self.ipget = ipget(resources=resources, metadata={"role": self.role})
 
                 # Initialize Lotus Kit if available and not disabled
                 if HAS_LOTUS and "lotus" not in disabled_components: