Container¶

class diwire.Container(*, autoregister: bool = True, autoregister_ignores: set[type[Any]] | None = None, autoregister_registration_factories: dict[type[Any], Callable[[Any], Registration]] | None = None, autoregister_default_lifetime: Lifetime = Lifetime.TRANSIENT, auto_compile: bool = True)[source]¶

Dependency injection container for registering and resolving services.

Supports automatic registration, lifetime singleton/transient, and factory patterns.

__init__(*, autoregister: bool = True, autoregister_ignores: set[type[Any]] | None = None, autoregister_registration_factories: dict[type[Any], Callable[[Any], Registration]] | None = None, autoregister_default_lifetime: Lifetime = Lifetime.TRANSIENT, auto_compile: bool = True) → None[source]¶

register(key: _C, /) → _C[source]¶

register(key: Callable[[...], T], /) → Callable[[...], T]

register(key: None = None, /, factory: None = None, instance: None = None, lifetime: Lifetime = Lifetime.TRANSIENT, scope: str | None = None, is_async: bool | None = None, concrete_class: type | None = None) → Callable[[T], T]

register(key: type, /, *, lifetime: Lifetime = Lifetime.TRANSIENT, scope: str | None = None, is_async: bool | None = None) → Callable[[T], T]

register(key: str, /, *, lifetime: Lifetime = Lifetime.TRANSIENT, scope: str | None = None, is_async: bool | None = None) → Callable[[T], T]

Can be used as: - Bare class decorator: @container.register - Parameterized decorator: @container.register(lifetime=Lifetime.SINGLETON) - Interface decorator: @container.register(IService) on a class - Factory function decorator: @container.register (with return annotation) - Direct call: container.register(IService, concrete_class=MyService)

Parameters:

key – The service key (interface/type) to register under. When None, returns a decorator. When used with @container.register(Interface) on a class, the decorated class becomes the implementation.
factory – Optional factory to create instances. Generator factories (Generator[T, None, None] or AsyncGenerator[T, None]) are supported for resource cleanup - the container calls close()/aclose() when the scope exits.
instance – Optional pre-created instance.
lifetime – The lifetime of the service. This default applies only to explicit registrations via register; auto-registration uses autoregister_default_lifetime from container configuration.
scope – Optional scope name for SCOPED services.
is_async – Whether the factory is async. If None, auto-detected from factory.
concrete_class – Optional concrete implementation class. When specified, key is used as the interface and concrete_class is the implementation.

Returns:

returns the class unchanged - When used as a parameterized decorator: returns a decorator function - When used as a direct call: returns None

Return type:

When used as a bare decorator on a class

Raises:

DIWireScopedWithoutScopeError – If lifetime is SCOPED but no scope is provided.
DIWireConcreteClassRequiresClassError – If concrete_class is not a class type.
DIWireDecoratorFactoryMissingReturnAnnotationError – If used as a factory decorator but the function has no return annotation and no explicit key.

Note

When using generator factories for cleanup, wrap cleanup code in try/finally:

def my_factory() -> Generator[Resource, None, None]:
    resource = acquire_resource()
    try:
        yield resource
    finally:
        resource.close()  # MUST be in finally block

Without try/finally, cleanup code after yield will not execute when the scope exits, as close()/aclose() raises GeneratorExit at the yield point.

enter_scope(scope_name: str | None = None) → ScopedContainer[source]¶

Start a new scope for resolving SCOPED dependencies.

The scope is activated immediately upon creation, allowing imperative usage:

scope = container.enter_scope(“request”) # … use the scope … scope.close() # or container.close() to close all scopes

Context manager usage is also supported:

with container.enter_scope(“request”) as scope:: # … use the scope …

Parameters:: scope_name – Optional name for the scope. If not provided, an integer ID is generated.
Returns:: A ScopedContainer that is already activated.

Note

Nested scopes inherit from parent scopes. A scope started within another scope will have access to dependencies registered for the parent scope.

compile() → None[source]¶

Compile all registered services into optimized providers.

This pre-compiles the dependency graph into specialized provider objects that eliminate runtime reflection and minimize dict lookups. Call this after all services have been registered for maximum performance.

resolve(key: None = None, *, scope: str) → Callable[[Callable[[...], Any]], Any][source]¶

resolve(key: None = None, *, scope: None = None) → Callable[[Callable[[...], Any]], Any]

resolve(key: type[T], *, scope: None = None) → T

resolve(key: type[T], *, scope: str) → T

resolve(key: Callable[[...], Coroutine[Any, Any, T]], *, scope: None = None) → _AsyncInjectedFunction[T]

resolve(key: Callable[[...], Coroutine[Any, Any, T]], *, scope: str) → _AsyncScopedInjectedFunction[T]

resolve(key: Callable[[...], T], *, scope: None = None) → _InjectedFunction[T]

resolve(key: Callable[[...], T], *, scope: str) → _ScopedInjectedFunction[T]

resolve(key: ServiceKey, *, scope: str | None = None) → Any

resolve(key: Any, *, scope: str | None = None) → Any

Resolve and return a service instance by its key.

When called with key=None, returns a decorator that can be applied to functions to enable dependency injection.

Parameters:

key – The service key to resolve. If None, returns a decorator.
scope – Optional scope name. If provided and key is a function, returns a ScopedInjected that creates a new scope per call.

Examples

# Direct usage:
injected = container.resolve(my_func, scope="request")


# Decorator usage:
@container.resolve(scope="request")
async def handler(service: Annotated[Service, Injected()]) -> dict: ...

async aresolve(key: type[T], *, scope: None = None) → T[source]¶

async aresolve(key: type[T], *, scope: str) → T

async aresolve(key: Callable[[...], Coroutine[Any, Any, T]], *, scope: None = None) → _AsyncInjectedFunction[T]

async aresolve(key: Callable[[...], Coroutine[Any, Any, T]], *, scope: str) → _AsyncScopedInjectedFunction[T]

async aresolve(key: ServiceKey, *, scope: str | None = None) → Any

async aresolve(key: Any, *, scope: str | None = None) → Any

Asynchronously resolve and return a service instance by its key.

This method supports async factories and async generator factories. Use this method when resolving services that have async dependencies.

Note

For decorator usage, use the synchronous .resolve() method which handles both sync and async functions correctly.

Parameters:

key – The service key to resolve.
scope – Optional scope name. If provided and key is a function, returns an AsyncScopedInjected that creates a new scope per call.

Raises:

DIWireAsyncGeneratorFactoryWithoutScopeError – If an async generator factory is used without an active scope.

Examples

# Direct usage: injected = await container.aresolve(my_func, scope=”request”)

close() → None[source]¶

Close all active scopes and mark the container as closed.

After calling this method, any attempt to resolve services or start new scopes will raise DIWireContainerClosedError.

Scopes are closed in LIFO order (newest first). This method is idempotent - calling it multiple times is safe.

If a scope’s close() fails, that scope remains in _active_scopes and the exception is re-raised.

async aclose() → None[source]¶

Asynchronously close all active scopes and mark the container as closed.

Use this method when scopes contain async generator factories that need proper async cleanup.

After calling this method, any attempt to resolve services or start new scopes will raise DIWireContainerClosedError.

Scopes are closed in LIFO order (newest first). This method is idempotent - calling it multiple times is safe.

This method will drain remaining scopes even if the container is already marked as closed. If a scope’s aclose() fails, that scope remains in _active_scopes and the exception is re-raised.

close_scope(scope_name: str) → None[source]¶

Close all active scopes that contain the given scope name.

This closes the named scope and all its child scopes in LIFO order (children first, then parents).

Parameters:: scope_name – The name of the scope to close.

Example

# Given hierarchy: app -> session -> request container.close_scope(“session”) # Closes both “request” and “session”

async aclose_scope(scope_name: str) → None[source]¶

Asynchronously close all active scopes that contain the given scope name.

This closes the named scope and all its child scopes in LIFO order (children first, then parents).

Parameters:: scope_name – The name of the scope to close.