container_context¶
container_context is an instance of an internal proxy class that forwards calls to the “current” container stored
in contextvars.
- class diwire.container_context._ContainerContextProxy[source]¶
Lazy proxy that forwards calls to the current container from context.
This allows setting up decorators before the container is configured, with the actual container lookup happening at call time.
Resolution order in get_current(): 1. ContextVar (highest precedence - for explicit per-request containers) 2. Thread-local (for asyncio.run() case where ContextVar doesn’t propagate) 3. Instance-level default (lowest precedence - for cross-thread access)
The instance-level default exists because some frameworks (e.g., FastAPI/Starlette) run sync endpoint handlers in a thread pool, meaning neither ContextVar nor thread-local storage can access a container set in the main thread.
Registrations can be deferred until a container is set; they are applied the next time set_current() is called.
- set_current(container: Container) Token[Container | None][source]¶
Set the current container in the context.
Sets the container in three storage mechanisms: 1. ContextVar - for same async context access 2. Thread-local - for asyncio.run() case (same thread, new context) 3. Instance-level default - for thread pool access (different threads)
- Parameters:
container – The container to set as current.
- Returns:
A token that can be used to reset the container.
- get_current() Container[source]¶
Get the current container from the context.
Resolution order (first non-None wins): 1. ContextVar - for per-context containers 2. Thread-local - for asyncio.run() (same thread, new context) 3. Instance-level default - for thread pools (different thread entirely)
- Returns:
The current container.
- Raises:
DIWireContainerNotSetError – If no container has been set.
- reset(token: Token[Container | None]) None[source]¶
Reset the container to its previous value.
- Parameters:
token – The token returned by set_current.
- 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[[...], Any], *, scope: None = None) Any
- resolve(key: Callable[[...], Any], *, scope: str) Any
- resolve(key: Any, *, scope: str | None = None) Any
Resolve a service or create a dependency-injected wrapper.
When called with key=None, returns a decorator that can be applied to functions to enable dependency injection with lazy container lookup.
When called with a type, resolves and returns a service instance from the current container.
- Parameters:
key – The service key to resolve, or None for decorator usage.
scope – Optional scope name for scoped resolution.
- Returns:
A service instance, or a wrapper for function decoration.
Examples
# Decorator usage (container looked up at call time): @container_context.resolve(scope="request") async def handler(service: Annotated[Service, Injected()]) -> dict: ... # Direct resolution: service = container_context.resolve(Service)
- aresolve(key: type[T], *, scope: str | None = None) Coroutine[Any, Any, T][source]¶
Asynchronously resolve a service from the current container.
- Parameters:
key – The service key to resolve.
scope – Optional scope name for scoped resolution.
- Returns:
A coroutine that resolves to the service instance.
- 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]
- register(key: Any, /, factory: type[FactoryClassProtocol] | Callable[[...], Any | Generator[Any, None, None]] | None = None, instance: Any | None = None, lifetime: Lifetime = Lifetime.TRANSIENT, scope: str | None = None, is_async: bool | None = None, concrete_class: type | None = None) None
Register a service with the current container.
Supports the same decorator and direct-call patterns as Container.register. If no container is set, registration is deferred until set_current().
- close() None[source]¶
Close the current container.
Closes all active scopes and marks the container as closed. After calling this method, any attempt to resolve services or start new scopes will raise DIWireContainerClosedError.
- async aclose() None[source]¶
Asynchronously close the current container.
Use this method when scopes contain async generator factories that need proper async cleanup.