Skip to content

di

Classes:

  • CircularDependencyError

    Raised when dependency resolution encounters a reference cycle.

  • DiContainer

    A lightweight configuration-based Dependency Injection (DI) container.

Functions:

  • load_class

    Dynamically load a class from a module path string.

CircularDependencyError


              flowchart TD
              oqtopus_util.di.CircularDependencyError[CircularDependencyError]

              

              click oqtopus_util.di.CircularDependencyError href "" "oqtopus_util.di.CircularDependencyError"
            

Raised when dependency resolution encounters a reference cycle.

DiContainer

DiContainer(registry: dict[str, Any])

A lightweight configuration-based Dependency Injection (DI) container.

The container receives a fully-parsed configuration dictionary (after environment-variable substitution via load_config()) and provides objects based on _target_ class paths.

Features
  • Supports _scope_ = "singleton" (default) or "prototype".
  • _target_ is required for every component.
    • String values starting with @ are treated as dependency references.
  • Keys starting with "_" (e.g., _target_, _scope_) are metadata and excluded from constructor arguments.
  • Uses Python's importlib to dynamically import target classes.
  • Singleton instances are cached inside the container.

Example YAML:

job_fetcher:
  _target_: oqtopus_engine_core.fetchers.OqtopusCloudJobFetcher
  _scope_: singleton
            repo: "@job_repo"

        job_repo:
            _target_: oqtopus_engine_core.repositories.JobRepository
            base_url: "http://localhost:8888"

Example usage:

dicon = DiContainer(registry_config)
job_fetcher = dicon.get("job_fetcher")

Parameters:

  • registry

    (dict[str, Any]) –

    Configuration dictionary for the dependency registry. Top-level keys represent dependency names.

Methods:

  • get

    Retrieve a dependency by name.

get

get(name: str) -> Any

Retrieve a dependency by name.

Behavior
  • Raise KeyError if the name does not exist in the config.
  • Import the target class defined by _target_.
  • Create an instance with kwargs from the component config.
  • Respect _scope_:
    • singleton (default): cache the instance
    • prototype: always create a fresh instance

Parameters:

  • name

    (str) –

    Component name to retrieve.

Returns:

  • Any

    The created or cached instance.

Raises:

  • KeyError

    If the component name is missing.

  • ValueError

    If _target_ is missing.

  • CircularDependencyError

    If dependency references form a cycle.

  • ImportError

    If module/class cannot be imported.

  • TypeError

    If constructor arguments mismatch.

load_class

load_class(path: str) -> type[Any]

Dynamically load a class from a module path string.

Parameters:

  • path

    (str) –

    Fully qualified class path (e.g. "mypackage.module.MyClass").

Returns:

  • type[Any]

    The class object referenced by the path.