Dependency Injection

The dependency injection module provides a robust and flexible dependency injection system for Python applications.

Decorators

retrosys.core.dependency_injection.decorators.injectable(lifecycle=Lifecycle.SINGLETON, context_key='', is_async=False, resolution_strategy=ResolutionStrategy.EAGER)[source]

Marks a class as injectable by the dependency injection container.

This decorator attaches special metadata attributes to a class to make it recognizable by the dependency injection system. It configures how the class should be instantiated, cached, and resolved by the container.

Parameters:

lifecycle (Lifecycle) – Determines how instances are created and cached.
context_key (str) – Optional key for contextual binding, allowing multiple implementations of the same type. Defaults to an empty string.
is_async (bool) – Whether this service requires asynchronous initialization.
resolution_strategy (ResolutionStrategy) – Whether to resolve the service eagerly or lazily.

Returns:

A decorator function that attaches DI metadata to the decorated class.

Side Effects:: Sets the following attributes on the decorated class: - __di_injectable__: Marker indicating the class is injectable - __di_lifecycle__: The lifecycle strategy - __di_context_key__: The context key for disambiguating registrations - __di_is_async__: Whether the class requires async initialization - __di_resolution_strategy__: The resolution strategy Also auto-detects async initialization based on the __init__ method.

retrosys.core.dependency_injection.decorators.inject_property(service_type)[source]

Creates a property that automatically resolves a dependency when accessed.

This decorator creates a descriptor that lazily resolves a dependency from the container when the property is accessed. It enables property injection by using a backing field to store the resolved instance.

Parameters:: service_type (Type) – The type of service to be injected when the property is accessed. This should be a type (class) that is registered with the container.
Returns:: A property descriptor that resolves and returns the dependency when accessed.

Example

class MyService:
    @inject_property(IDatabase)
    def database(self):
        pass  # The method body is not used

retrosys.core.dependency_injection.decorators.inject_method(params)[source]

Automatically injects dependencies as parameters to a method.

This decorator allows method parameters to be automatically injected from the dependency injection container. It wraps the original method and resolves the specified dependencies before calling it, unless they are explicitly provided in the method call.

Parameters:: params (Dict[str, Type]) – A dictionary mapping parameter names to their types. Each entry specifies a parameter to be injected and the type of service to resolve from the container.
Returns:: A decorated method that automatically resolves and injects dependencies.

Example

class MyService:
    @inject_method({'database': IDatabase, 'logger': ILogger})
    def process_data(self, data, database, logger):
        # database and logger are automatically injected
        pass

retrosys.core.dependency_injection.decorators.register_module(container)[source]

Scans a class for injectable components and registers them with a container.

This decorator examines a class (typically a module definition class) and finds all of its member classes that have been marked as injectable using the @injectable decorator. It creates a Module instance and registers all discovered injectable components with that module, then registers the module with the provided container.

Parameters:: container – The dependency injection container to register the module with. This should be an instance of Container.
Returns:: A decorator function that processes a class and returns it after registering all its injectable members.
Raises:: TypeError – If the decorated object is not a class.

Example

@register_module(container)
class DataModule:
    @injectable()
    class DatabaseService:
        pass

    @injectable()
    class RepositoryService:
        pass

Container

class retrosys.core.dependency_injection.container.Container[source]

Bases: object

Main dependency injection container with both synchronous and asynchronous support.

The Container is the central registry and resolution mechanism for the dependency injection system. It manages service registrations, handles instantiation of dependencies according to their lifecycle policies, resolves dependencies for constructor/property/method injection, and provides support for hierarchical containers, scopes, and modules.

Features:

Support for singleton, transient, and scoped service lifetimes
Constructor, property, and method injection
Synchronous and asynchronous dependency resolution
Lazy dependency resolution
Module-based organization
Child containers for isolated dependency graphs
Testing support with mock services

_descriptors: Dictionary mapping service types to lists of service descriptors

_resolution_stack: Stack used for detecting circular dependencies during resolution

_lock: Thread lock for thread-safety

_modules: Dictionary of registered modules by namespace

_logger: Logger for container events

_test_mode: Flag indicating whether test mode is enabled

_mock_instances: Dictionary of mock instances used in test mode

_signature_cache: Cache of constructor signatures for performance

_property_injection_cache: Cache of property injection metadata

_method_injection_cache: Cache of method injection metadata

register(service_type, implementation_type=None, lifecycle=Lifecycle.SINGLETON, factory=None, context_key='', is_async=False, resolution_strategy=ResolutionStrategy.EAGER, on_init=None, on_destroy=None)[source]

Register a service with the container.

This method registers a service type with its implementation type, lifecycle, factory functions, and other configuration options in the container. This is the primary method for configuring dependencies in the dependency injection system.

Parameters:

service_type (Type[TypeVar(T)]) – The type to register, typically an interface or abstract class.
implementation_type (Optional[Type]) – The concrete implementation type to instantiate when resolving the service_type. If None, the service_type itself is used as the implementation.
lifecycle (Lifecycle) – Determines how instances are created and cached (singleton, transient, scoped). Defaults to Lifecycle.SINGLETON.
factory (Union[Callable[[ContainerProtocol], Any], Callable[[ContainerProtocol], Awaitable[Any]], None]) – Optional factory function to create instances of the service. If provided, this is used instead of constructor injection.
context_key (str) – Optional key for contextual binding, allowing multiple implementations of the same type to be registered with different keys.
is_async (bool) – Whether this service requires asynchronous initialization. Set to True for services that have async dependencies or initialization logic.
resolution_strategy (ResolutionStrategy) – Whether to resolve the service eagerly or lazily. Defaults to ResolutionStrategy.EAGER.
on_init (Optional[Callable[[Any], Optional[Awaitable[None]]]]) – Optional callback function to invoke after a service instance is created.
on_destroy (Optional[Callable[[Any], Optional[Awaitable[None]]]]) – Optional callback function to invoke when a service instance is being destroyed.

Returns:

The container instance for method chaining.

Return type:

Container

Side Effects:

Creates a ServiceDescriptor and adds it to the container’s registry.
Updates property and method injection caches if applicable.

register_instance(service_type, instance)[source]

Register an existing instance with the container.

This is a convenience method for registering pre-constructed instances as singletons. The instance is registered with the container and will be returned directly on resolution without creating a new instance.

Parameters:

service_type (Type[TypeVar(T)]) – The type to register the instance as, typically an interface or base class.
instance (TypeVar(T)) – The pre-constructed instance to register.

Returns:

The container instance for method chaining.

Return type:

Container

Side Effects:: Creates a singleton ServiceDescriptor with the provided instance and adds it to the registry.

register_factory(service_type, factory, lifecycle=Lifecycle.SINGLETON, is_async=False, context_key='')[source]

Register a factory function for a service.

This is a convenience method for registering a factory function that creates service instances. The factory function receives the container as a parameter and returns an instance of the service.

Parameters:

service_type (Type[TypeVar(T)]) – The type to register, typically an interface or abstract class.
factory (Union[Callable[[ContainerProtocol], Any], Callable[[ContainerProtocol], Awaitable[Any]]]) – A function that creates an instance of the service. For async factories, this should return a coroutine.
lifecycle (Lifecycle) – Determines how instances are created and cached. Defaults to Lifecycle.SINGLETON.
is_async (bool) – Whether this factory is asynchronous. This is automatically detected for coroutine functions but can be explicitly set.
context_key (str) – Optional key for contextual binding.

Returns:

The container instance for method chaining.

Return type:

Container

Side Effects:: Automatically detects if the factory is a coroutine function and sets is_async accordingly.

lazy_resolve(service_type, context_key='')[source]

Get a lazy wrapper for a dependency.

This method returns a proxy object that delays the actual resolution of the service until it is first accessed. This is useful for breaking circular dependencies and for performance optimization when a service might not be used.

Parameters:

service_type (Type[TypeVar(T)]) – The type of service to lazily resolve.
context_key (str) – Optional key for contextual binding. Defaults to an empty string.

Returns:

A lazy proxy object that will resolve the service when accessed.

Return type:

Lazy[T]

resolve(service_type, context_key='')[source]

Synchronously resolve a service from the container.

This method resolves and returns an instance of the requested service type. It handles constructor injection, property injection, and method injection, and manages caching of singleton instances.

Parameters:

service_type (Type[TypeVar(T)]) – The type of service to resolve.
context_key (str) – Optional key for contextual binding. Defaults to an empty string.

Returns:

An instance of the requested service type.

Return type:

T

Raises:

CircularDependencyError – If a circular dependency is detected during resolution.
DependencyNotFoundError – If the service type is not registered or cannot be resolved.
AsyncInitializationError – If attempting to synchronously resolve an async service.

Side Effects:

Creates and caches instances for singleton services.
Calls on_init lifecycle hooks for newly created instances.

async resolve_async(service_type, context_key='')[source]

Asynchronously resolve a service from the container.

This method resolves and returns an instance of the requested service type, supporting asynchronous initialization and dependencies. It handles constructor injection, property injection, and method injection for async services.

Parameters:

service_type (Type[TypeVar(T)]) – The type of service to resolve.
context_key (str) – Optional key for contextual binding. Defaults to an empty string.

Returns:

An instance of the requested service type.

Return type:

T

Raises:

CircularDependencyError – If a circular dependency is detected during resolution.
DependencyNotFoundError – If the service type is not registered or cannot be resolved.

Side Effects:

Creates and caches instances for singleton services.
Calls on_init lifecycle hooks for newly created instances.
Awaits async factories and async on_init hooks.

create_scope()[source]

Create a new dependency scope.

A scope provides a controlled lifetime for scoped services and manages their disposal. Services registered with Lifecycle.SCOPED will be instantiated once per scope.

Returns:: A new scope object that inherits registrations from this container.
Return type:: Scope

register_module(module, namespace='')[source]

Register a module with the container.

Modules provide a way to organize related services and their dependencies. When a module is registered, all its service registrations are imported into the container.

Parameters:

module (Module) – The module instance to register.
namespace (str) – Optional namespace to use for the module. If not provided, the module’s name attribute will be used.

Returns:

The container instance for method chaining.

Return type:

Container

Side Effects:

Adds the module to the container’s module registry.
Imports all service registrations from the module.
Issues a warning if a module with the same namespace is already registered.

create_child_container()[source]

Create a new container that inherits registrations from this one.

A child container inherits all service registrations from its parent container, but can have its own registrations that override the parent’s. This allows for creating isolated dependency graphs that still have access to shared services.

Returns:: A new container instance that inherits registrations from this container.
Return type:: Container

Side Effects:: Copies all registrations and module references from the parent to the child container.

enable_test_mode()[source]

Enable test mode for mocking dependencies.

In test mode, the container will check for mock instances before attempting normal resolution. This allows for easy replacement of real services with test doubles during unit testing.

Returns:: The container instance for method chaining.
Return type:: Container

Side Effects:: Sets the test_mode flag to True.

disable_test_mode()[source]

Disable test mode and clear all registered mocks.

This method turns off test mode and removes all mock instances that were registered with the container. After calling this method, the container will return to normal resolution behavior.

Returns:: The container instance for method chaining.
Return type:: Container

Side Effects:

Sets the test_mode flag to False.
Clears the mock_instances dictionary.

mock(service_type, instance)[source]

Register a mock instance for testing.

This method is used in test mode to replace a registered service with a mock implementation. When the specified service_type is resolved, the mock instance will be returned instead of creating a new instance or using the regular singleton.

Parameters:

service_type (Type[TypeVar(T)]) – The service type to mock.
instance (TypeVar(T)) – The mock instance to use when resolving the service_type.

Returns:

The container instance for method chaining.

Return type:

Container

Side Effects:: Adds the mock instance to the _mock_instances dictionary.

async dispose()[source]

Dispose of all services with on_destroy handlers.

This method is responsible for cleaning up singleton instances that have on_destroy lifecycle hooks. It should be called when the container is no longer needed to properly release resources held by services.

For each singleton service with an on_destroy handler, this method will: - Call the on_destroy handler synchronously if it’s not async - Await the on_destroy handler if it’s async

Return type:: None
Returns:: None

Side Effects:: Invokes on_destroy handlers for singleton instances, which may release resources, close connections, or perform other cleanup operations.

Module

class retrosys.core.dependency_injection.module.Module(name='')[source]

Bases: object

A group of related dependencies in the dependency injection system.

Modules provide a way to organize related services into logical groups and encapsulate their registrations. They act as containers for related dependencies and can be registered with a parent container to make all their services available.

Modules have their own container for registering services, but can also access services from their parent container when registered with one.

_container: The internal container used for service registrations.

parent_container: Reference to the parent container if registered.

name: The name of the module, used for identification and namespace.

Initialize a new module.

Creates a new module with its own internal container for service registrations.

Parameters:: name (str) – Optional name for the module, used for identification when registered with a parent container. Defaults to an empty string.

__init__(name='')[source]

Initialize a new module.

Creates a new module with its own internal container for service registrations.

Parameters:: name (str) – Optional name for the module, used for identification when registered with a parent container. Defaults to an empty string.

register(service_type, implementation_type=None, lifecycle=Lifecycle.SINGLETON, factory=None, context_key='', is_async=False, resolution_strategy=ResolutionStrategy.EAGER, on_init=None, on_destroy=None)[source]

Register a service with the module.

This method registers a service type with its implementation type, lifecycle, factory functions, and other configuration options in the module’s container.

Parameters:

service_type (Type[TypeVar(T)]) – The type to register, typically an interface or abstract class.
implementation_type (Optional[Type]) – The concrete implementation type to instantiate when resolving the service_type. If None, the service_type itself is used as the implementation.
lifecycle (Lifecycle) – Determines how instances are created and cached (singleton, transient, scoped). Defaults to Lifecycle.SINGLETON.
factory (Union[Callable[[ContainerProtocol], Any], Callable[[ContainerProtocol], Awaitable[Any]], None]) – Optional factory function to create instances of the service. If provided, this is used instead of constructor injection.
context_key (str) – Optional key for contextual binding, allowing multiple implementations of the same type to be registered with different keys.
is_async (bool) – Whether this service requires asynchronous initialization. Set to True for services that have async dependencies or initialization logic.
resolution_strategy (ResolutionStrategy) – Whether to resolve the service eagerly or lazily. Defaults to ResolutionStrategy.EAGER.
on_init (Optional[Callable[[Any], Optional[Awaitable[None]]]]) – Optional callback function to invoke after a service instance is created.
on_destroy (Optional[Callable[[Any], Optional[Awaitable[None]]]]) – Optional callback function to invoke when a service instance is being destroyed.

Returns:

The module instance for method chaining.

Return type:

Module

register_instance(service_type, instance)[source]

Register an existing instance with the module.

This is a convenience method for registering pre-constructed instances as singletons within the module’s container.

Parameters:

service_type (Type[TypeVar(T)]) – The type to register the instance as, typically an interface or base class.
instance (TypeVar(T)) – The pre-constructed instance to register.

Returns:

The module instance for method chaining.

Return type:

Module

register_factory(service_type, factory, lifecycle=Lifecycle.SINGLETON, is_async=False)[source]

Register a factory function for a service with the module.

This is a convenience method for registering a factory function that creates service instances within the module’s container.

Parameters:

service_type (Type[TypeVar(T)]) – The type to register, typically an interface or abstract class.
factory (Union[Callable[[ContainerProtocol], Any], Callable[[ContainerProtocol], Awaitable[Any]]]) – A function that creates an instance of the service. For async factories, this should return a coroutine.
lifecycle (Lifecycle) – Determines how instances are created and cached. Defaults to Lifecycle.SINGLETON.
is_async (bool) – Whether this factory is asynchronous. This is automatically detected for coroutine functions but can be explicitly set.

Returns:

The module instance for method chaining.

Return type:

Module

Side Effects:: Automatically detects if the factory is a coroutine function and sets is_async accordingly.

resolve(service_type, context_key='')[source]

Resolve a service from the module.

This method resolves and returns an instance of the requested service type. It first checks the module’s own container, and if not found there, falls back to the parent container (if registered with one).

Parameters:

service_type (Type[TypeVar(T)]) – The type of service to resolve.
context_key (str) – Optional key for contextual binding. Defaults to an empty string.

Returns:

An instance of the requested service type.

Return type:

T

Raises:

DependencyNotFoundError – If the service type is not registered in this module or its parent container.

async resolve_async(service_type, context_key='')[source]

Asynchronously resolve a service from the module.

This method asynchronously resolves and returns an instance of the requested service type. It first checks the module’s own container, and if not found there, falls back to the parent container (if registered with one).

Parameters:

service_type (Type[TypeVar(T)]) – The type of service to resolve.
context_key (str) – Optional key for contextual binding. Defaults to an empty string.

Returns:

An instance of the requested service type.

Return type:

T

Raises:

DependencyNotFoundError – If the service type is not registered in this module or its parent container.

Service Descriptor

class retrosys.core.dependency_injection.service_descriptor.ServiceDescriptor(service_type, implementation_type=None, lifecycle=Lifecycle.SINGLETON, factory=None, instance=None, context_key='', is_async=False, resolution_strategy=ResolutionStrategy.EAGER, metadata=<factory>, on_init=None, on_destroy=None, property_injections=<factory>, method_injections=<factory>)[source]

Bases: object

Describes a registered service.

Parameters:

service_type (Type)
implementation_type (Type | None)
lifecycle (Lifecycle)
factory (Callable[[ContainerProtocol], Any] | Callable[[ContainerProtocol], Awaitable[Any]] | None)
instance (Any)
context_key (str)
is_async (bool)
resolution_strategy (ResolutionStrategy)
metadata (Dict[str, Any])
on_init (Callable[[Any], Awaitable[None] | None] | None)
on_destroy (Callable[[Any], Awaitable[None] | None] | None)
property_injections (Dict[str, Type])
method_injections (Dict[str, Dict[str, Type]])

service_type: Type

implementation_type: Optional[Type] = None

lifecycle: Lifecycle = 1

factory: Union[Callable[[ContainerProtocol], Any], Callable[[ContainerProtocol], Awaitable[Any]], None] = None

instance: Any = None

context_key: str = ''

is_async: bool = False

resolution_strategy: ResolutionStrategy = 1

metadata: Dict[str, Any]

on_init: Optional[Callable[[Any], Optional[Awaitable[None]]]] = None

on_destroy: Optional[Callable[[Any], Optional[Awaitable[None]]]] = None

property_injections: Dict[str, Type]

method_injections: Dict[str, Dict[str, Type]]

is_resolved()[source]

Determines if the service instance has been resolved/initialized.

This method checks whether a service has been instantiated based on its lifecycle. For singleton services, it verifies if the instance attribute is populated. For non-singleton services, it always returns False since they are created on-demand and not cached.

Returns:

True if the service is a singleton and has been instantiated,: False otherwise.

Return type:

bool

Scope

class retrosys.core.dependency_injection.scope.Scope(parent_container)[source]

Bases: object

Represents a dependency injection scope.

Initialize a new dependency injection scope.

Creates a new scope with its own container that inherits from the parent container. The scope maintains a cache of resolved service instances.

Parameters:: parent_container (ContainerProtocol) – The parent container that this scope will create a child container from.

Side Effects:: Creates a child container from the parent container.

__init__(parent_container)[source]

Initialize a new dependency injection scope.

Creates a new scope with its own container that inherits from the parent container. The scope maintains a cache of resolved service instances.

Parameters:: parent_container (ContainerProtocol) – The parent container that this scope will create a child container from.

Side Effects:: Creates a child container from the parent container.

resolve(service_type, context_key='')[source]

Resolves a service instance within the current scope.

This method retrieves or creates a service instance of the specified type. If the instance already exists in the scope’s cache, it returns the cached instance. Otherwise, it resolves the service from the container and caches it if it has a scoped lifecycle.

Parameters:

service_type (Type[T]) – The type of service to resolve.
context_key (str, optional) – A key for contextual binding resolution. Defaults to an empty string.

Returns:

An instance of the requested service type.

Return type:

T

Side Effects:: Updates the internal instances cache if the resolved service has a scoped lifecycle.

async resolve_async(service_type, context_key='')[source]

Asynchronously resolves a service instance within the current scope.

This method retrieves or creates a service instance of the specified type. If the instance already exists in the scope’s cache, it returns the cached instance. Otherwise, it asynchronously resolves the service from the container and caches it if it has a scoped lifecycle.

Parameters:

service_type (Type[T]) – The type of service to resolve.
context_key (str, optional) – A key for contextual binding resolution. Defaults to an empty string.

Returns:

An instance of the requested service type.

Return type:

T

Side Effects:: Updates the internal instances cache if the resolved service has a scoped lifecycle.

async __aenter__()[source]

Async context manager entry point.

Allows the scope to be used as an async context manager with the ‘async with’ statement.

Returns:: Returns self to be used within the context manager.
Return type:: Scope

async __aexit__(exc_type, exc_val, exc_tb)[source]

Async context manager exit point.

Called when exiting an ‘async with’ block. Automatically disposes of all scoped instances by calling the dispose method.

Parameters:

exc_type – The exception type if an exception was raised in the context block, None otherwise.
exc_val – The exception value if an exception was raised in the context block, None otherwise.
exc_tb – The traceback if an exception was raised in the context block, None otherwise.

Side Effects:: Calls the dispose method to clean up all scoped instances.

async dispose()[source]

Dispose of all scoped instances.

This method cleans up all service instances managed by this scope. It attempts to call the dispose method on instances if they have one, or calls the on_destroy handler if specified in the service descriptor. Finally, it clears the instances cache.

The method handles both synchronous and asynchronous dispose/on_destroy methods.

Side Effects:

Calls dispose methods on service instances if available
Calls on_destroy handlers from service descriptors if available
Catches and logs any exceptions during disposal
Clears the internal instances cache

Lazy Resolution

class retrosys.core.dependency_injection.lazy.Lazy(container, service_type, context_key='')[source]

Bases: Generic[T]

Wrapper for lazy dependency resolution.

This class provides a way to delay the resolution of a dependency until it is actually needed. It acts as a proxy that resolves the dependency only when accessed, which can help break circular dependencies and improve performance when expensive dependencies might not always be used.

_container: The dependency injection container used to resolve the service.

_service_type: The type of service to be lazily resolved.

_context_key: Optional key for contextual binding.

_instance: Cached instance of the resolved service (None until resolved).

_resolved: Flag indicating whether the service has been resolved.

Usage:

# As a variable lazy_service = container.lazy_resolve(ServiceType) # Later, when needed service_instance = lazy_service()

# For async resolution service_instance = await lazy_service.async_resolve()

Initialize a new lazy dependency reference.

Parameters:

container (ContainerProtocol) – The container that will be used to resolve the service.
service_type (Type[TypeVar(T)]) – The type of service to be lazily resolved.
context_key (str) – Optional key for contextual binding. Defaults to an empty string.

__init__(container, service_type, context_key='')[source]

Initialize a new lazy dependency reference.

Parameters:

container (ContainerProtocol) – The container that will be used to resolve the service.
service_type (Type[TypeVar(T)]) – The type of service to be lazily resolved.
context_key (str) – Optional key for contextual binding. Defaults to an empty string.

__call__()[source]

Resolve the dependency when the lazy object is called.

This method allows the Lazy object to be used like a function. When called, it resolves the dependency if not already resolved, caches the instance, and returns it.

Returns:: The resolved service instance.
Return type:: T

Side Effects:: Resolves the dependency and caches it on first call.

async async_resolve()[source]

Asynchronously resolve the dependency.

This method is the asynchronous counterpart to __call__. It should be used when working with services that require async resolution.

Returns:: The resolved service instance.
Return type:: T

Side Effects:: Asynchronously resolves the dependency and caches it on first call.

Project Types

class retrosys.core.dependency_injection.project_types.Lifecycle(value)[source]

Bases: Enum

Defines how dependencies are instantiated and cached.

This enum specifies the lifecycle strategy for dependency instances within the container, determining when instances are created and how long they are retained.

SINGLETON: One instance per container. The instance is created once and reused for all resolutions within the container’s lifetime.

TRANSIENT: New instance per resolution. Every time the dependency is requested, a new instance is created.

SCOPED: One instance per scope (e.g., request). The instance is created once per defined scope and reused within that scope.

SINGLETON = 1

TRANSIENT = 2

SCOPED = 3

class retrosys.core.dependency_injection.project_types.ResolutionStrategy(value)[source]

Bases: Enum

Defines when dependencies are resolved in the container.

This enum specifies the strategy for when dependencies should be resolved, either immediately upon registration or delayed until the dependency is requested.

EAGER: Resolve immediately when the dependency is registered in the container. This front-loads initialization costs but ensures dependencies are valid early.

LAZY: Resolve only when the dependency is actually requested from the container. This defers initialization costs but may delay discovery of configuration issues.

EAGER = 1

LAZY = 2

class retrosys.core.dependency_injection.project_types.ContainerProtocol(*args, **kwargs)[source]

Bases: Protocol

Protocol defining the interface for a dependency injection container.

This protocol specifies the methods that any dependency injection container must implement to properly resolve dependencies and manage container hierarchies.

The container is responsible for resolving services based on their type and optional context keys, handling both synchronous and asynchronous resolution.

resolve(service_type, context_key='')[source]

Resolves a service instance from the container.

Parameters:

service_type (Type[TypeVar(T)]) – The type of service to resolve.
context_key (str) – Optional context key for disambiguating multiple registrations of the same type. Defaults to an empty string.

Returns:

An instance of the requested service type.

Return type:

T

Raises:

ResolutionError – If the service cannot be resolved or has not been registered.

async resolve_async(service_type, context_key='')[source]

Asynchronously resolves a service instance from the container.

Parameters:

service_type (Type[TypeVar(T)]) – The type of service to resolve.
context_key (str) – Optional context key for disambiguating multiple registrations of the same type. Defaults to an empty string.

Returns:

An instance of the requested service type.

Return type:

T

Raises:

ResolutionError – If the service cannot be resolved or has not been registered.

create_child_container()[source]

Creates a new child container with this container as parent.

Child containers inherit all registrations from their parent but can override them with their own registrations. Resolution in a child container checks the child first, then falls back to the parent if the service is not found.

Returns:: A new child container instance.
Return type:: ContainerProtocol

retrosys.core.dependency_injection.project_types.FactoryCallable

Type for factory functions that create service instances.

A function that accepts a container instance and returns a service instance. This is used when registering factory methods with the dependency injection container.

alias of Callable[[ContainerProtocol], Any]

retrosys.core.dependency_injection.project_types.AsyncFactoryCallable

Type for asynchronous factory functions that create service instances.

A function that accepts a container instance and returns a coroutine that resolves to a service instance. This is used when registering async factory methods with the dependency injection container.

alias of Callable[[ContainerProtocol], Awaitable[Any]]

Errors

exception retrosys.core.dependency_injection.errors.CircularDependencyError[source]

Bases: Exception

Raised when a circular dependency is detected in the container.

This exception is thrown during service resolution when the dependency injection container detects a circular reference in the dependency graph. For example, if Service A depends on Service B, which depends on Service C, which depends on Service A, this creates a circular dependency that cannot be resolved.

exception retrosys.core.dependency_injection.errors.DependencyNotFoundError(message, service_type=None, context_key=None)[source]

Bases: Exception

Raised when a requested dependency cannot be resolved from the container.

This exception occurs when attempting to resolve a service that hasn’t been registered in the container, or when the registered service cannot be properly instantiated due to missing dependencies.

service_type: The type of the service that could not be resolved.

context_key: The context key used in the resolution attempt, if any.

message: The base error message, which will be enhanced with service type and context key information.

Initialize a new DependencyNotFoundError with detailed information.

Parameters:

message – The base error message.
service_type – The type of service that could not be resolved.
context_key – The context key used in the resolution attempt, if any.

__init__(message, service_type=None, context_key=None)[source]

Initialize a new DependencyNotFoundError with detailed information.

Parameters:

message – The base error message.
service_type – The type of service that could not be resolved.
context_key – The context key used in the resolution attempt, if any.

exception retrosys.core.dependency_injection.errors.AsyncInitializationError[source]

Bases: Exception

Raised when asynchronous service initialization fails.

This exception is thrown when an error occurs during the asynchronous initialization of a service, such as when an async factory method fails or when an async on_init lifecycle hook throws an exception.