FrozenContainer

FrozenContainer

Immutable, validated view of a dependency graph.

Produced by Container.freeze(). Resolve values by key with resolve() / aresolve() or the frozen[key] shorthand, scope short-lived values with scope() / ascope(), wire functions with inject(), and substitute providers in tests with override().

The container registers no new providers; it holds only the singleton cache and the root scope. It is safe to share across threads and tasks: scopes are tracked per contextvars.Context, so concurrent requests never see each other's scoped instances, and construction of a cached provider is single-flighted across threads and event loops — a singleton is built exactly once no matter how many threads or tasks race for it.

Example

>>> from depin import Container
>>> class Greeter:
...     def hello(self) -> str:
...         return 'hi'
>>> di = Container().bind(Greeter).freeze()
>>> di[Greeter].hello()
'hi'

__getitem__

__getitem__(key: type[T] | Token[T]) -> T

Resolve key synchronously; shorthand for resolve().

Example

>>> from depin import Container, Token
>>> port = Token[int]('port')
>>> di = Container().value(port, 8080).freeze()
>>> di[port]
8080

resolve

resolve(
    key: type[T] | Token[T], *, tag: str | None = None
) -> T

Resolve a value by key, synchronously.

Returns the cached singleton or scoped instance if present, otherwise builds it (resolving its dependencies first). Synchronous resolution cannot drive async providers: if key or anything it depends on is async, this raises rather than blocking an event loop — use aresolve() instead.

Parameters:

Name	Type	Description	Default
key	type[T] | Token[T]	A class or Token to resolve.	required
tag	str | None	Selects among providers registered under key with a tag.	None

Raises:

Type	Description
MissingProviderError	No provider is registered for key / tag.
AsyncInSyncContextError	The provider, or a dependency, is async.
OutsideScopeError	key is scoped and no scope is active.
CircularDependencyError	This context re-enters construction of the same cached provider.

Example

>>> from depin import Container
>>> from depin.errors import MissingProviderError
>>> di = Container().freeze()
>>> try:
...     di.resolve(int)
... except MissingProviderError as exc:
...     print(exc)
no provider for int (tag=None)

aresolve async

aresolve(
    key: type[T] | Token[T], *, tag: str | None = None
) -> T

Resolve a value by key, asynchronously.

The async counterpart to resolve(). Handles both sync and async providers, awaiting async factories, async generators, and async context managers. Concurrent resolutions of the same cached provider are single-flighted, so a singleton or scoped value is built exactly once even under concurrency.

Parameters:

Name	Type	Description	Default
key	type[T] | Token[T]	A class or Token to resolve.	required
tag	str | None	Selects among providers registered under key with a tag.	None

Raises:

Type	Description
MissingProviderError	No provider is registered for key / tag.
OutsideScopeError	key is scoped and no scope is active.
CircularDependencyError	This task re-enters construction of the same cached provider.

scope

scope() -> Generator[ScopeFrame]

Open a synchronous scope for scoped providers and their teardown.

Scoped providers resolved inside the with block are built once for the block and cached in the yielded frame. On exit, their teardowns run in reverse order of construction; if several fail, the errors are collected into an ExceptionGroup so no failure hides another. Singletons are unaffected — they live on the container, not the scope.

Scopes nest: a scoped value built in an outer scope is reused inside a nested scope, not rebuilt. Open sibling scopes for independent instances. Use ascope() when any provider in the scope is async.

Raises:

Type	Description
ExceptionGroup	One or more teardowns failed. Protocol violations, including an async teardown in this synchronous scope, appear as TeardownError members of the group.

Example

>>> from collections.abc import Generator
>>> from depin import Container, Scope
>>> events: list[str] = []
>>> class Conn: ...
>>> def connect() -> Generator[Conn]:
...     events.append('open')
...     yield Conn()
...     events.append('close')
>>> di = Container().bind(connect, scope=Scope.SCOPED).freeze()
>>> with di.scope():
...     conn = di.resolve(Conn)
>>> events
['open', 'close']

ascope async

ascope() -> AsyncGenerator[ScopeFrame]

Open an asynchronous scope; the async counterpart to scope().

Required when scoped providers are async (async factories, async generators, async context managers). Teardowns — sync and async alike — run in reverse order on exit, with failures collected into an ExceptionGroup. The per-request scope opened by the FastAPI integration is an ascope.

Raises:

Type	Description
ExceptionGroup	One or more teardowns failed. A generator provider that violates its teardown protocol appears as a TeardownError member of the group.

close

close() -> None

Tear down singleton providers that own lifecycle resources, synchronously.

Drains the root scope, running the teardown half of every singleton generator / context-manager provider in reverse order of construction. Call this once on shutdown of a synchronous application. Scoped providers do not need it — they are drained when their scope() block exits. Failures are collected into an ExceptionGroup.

Raises:

Type	Description
AsyncInSyncContextError	A singleton requires an async teardown; use aclose() instead.
ExceptionGroup	One or more synchronous teardowns failed. A generator provider that violates its teardown protocol appears as a TeardownError member of the group.

Example

>>> from collections.abc import Generator
>>> from depin import Container
>>> events: list[str] = []
>>> class Pool: ...
>>> def pool() -> Generator[Pool]:
...     events.append('open')
...     yield Pool()
...     events.append('close')
>>> di = Container().bind(pool).freeze()
>>> _ = di[Pool]
>>> di.close()
>>> events
['open', 'close']

aclose async

aclose() -> None

Tear down singleton providers that own lifecycle resources, asynchronously.

The async counterpart to close(), and the one to call when any singleton is an async provider. Drains the root scope in reverse order of construction, collecting failures into an ExceptionGroup.

Raises:

Type	Description
ExceptionGroup	One or more teardowns failed. A generator provider that violates its teardown protocol appears as a TeardownError member of the group.

reset

reset() -> None

Tear down every built singleton and drop the cache, so the next resolution rebuilds.

The difference from close() is what happens afterwards: close() drains the singletons and leaves them cached, while reset() drops them, so the container is usable again and builds fresh values on demand. Scoped and transient providers are untouched — a scoped value belongs to its scope, and a transient one is never cached.

Primarily a testing seam: it is what makes an override() reach a consumer that was already built, and it is what the depin.ext.pytest fixtures use. Do not call it while another thread or task may be resolving through this container — it drops the cache without coordinating with an in-flight construction, so one racing with a resolution can hand a caller a value whose teardown already ran.

Raises:

Type	Description
ExceptionGroup	One or more teardowns failed. Every failure is reported, and the cache is dropped either way. Protocol violations, including an async teardown in this synchronous drain, appear as TeardownError members; use areset() for async providers.

Example

>>> from depin import Container
>>> class Clock: ...
>>> di = Container().bind(Clock).freeze()
>>> first = di[Clock]
>>> di.reset()
>>> di[Clock] is first
False

areset async

areset() -> None

Tear down every built singleton and drop the cache; the counterpart to reset().

The one to call when any singleton is an async provider. Otherwise identical, including the same caveat about calling it while another thread or task may be resolving through this container.

Raises:

Type	Description
ExceptionGroup	One or more teardowns failed. Every failure is reported, and the cache is dropped either way. A generator provider that violates its teardown protocol appears as a TeardownError member of the group.

warmup

warmup() -> WarmupReport

Construct every singleton now, instead of on first resolution.

Walks the plan in resolution order, building each singleton that is not built already, so a provider that fails does so at startup rather than on the first request that needs it. Scoped and transient providers are untouched: a scoped value belongs to a scope, and a transient one is never cached. Calling it twice constructs nothing the second time.

A failure propagates unchanged — a container with some singletons built and one failed is a startup to abort, not a state to report.

Raises:

Type	Description
AsyncInSyncContextError	Some singleton needs async resolution. Nothing is constructed before this is raised; use awarmup().

Example

>>> from depin import Container
>>> class Config: ...
>>> class Service:
...     def __init__(self, config: Config) -> None: ...
>>> di = Container().bind(Config).bind(Service).freeze()
>>> report = di.warmup()
>>> len(report.constructed), len(report.cached)
(2, 0)

awarmup async

awarmup() -> WarmupReport

Construct every singleton now; the async counterpart to warmup().

Drives async singletons as well as sync ones, so it is what an ASGI lifespan calls. Otherwise identical: resolution order, the same report, and a failure that propagates unchanged.

Raises:

Type	Description
CircularDependencyError	This task re-enters construction of the same cached provider.

checks

checks() -> tuple[HealthCheck, ...]

Return the verification callables the bindings declared, as data.

Resolves nothing and runs nothing: this is the declaration, in resolution order. health() and ahealth() are what run them.

Example

>>> from depin import Container
>>> class Database: ...
>>> def ping(db: Database) -> None: ...
>>> di = Container().bind(Database, check=ping).freeze()
>>> [check.key.__qualname__ for check in di.checks()]
['Database']

health

health() -> HealthReport

Run every declared check and report what each said.

Each check receives the value its provider resolves to, and is healthy unless it raises or returns False. Every check runs: one failure never hides another, and a raised exception is carried on its HealthResult rather than propagating. An error raised while resolving a provider does propagate — a container that cannot build a provider is misused, not unhealthy.

Raises:

Type	Description
AsyncInSyncContextError	Some check needs an event loop, because its provider is async or the check is. Nothing runs before this is raised; use ahealth().
InvalidProviderError	A check returned an awaitable.
OutsideScopeError	A check's provider is scoped and no scope is active.

Example

>>> from depin import Container
>>> class Database:
...     ready = False
>>> def ping(db: Database) -> bool:
...     return db.ready
>>> di = Container().bind(Database, check=ping).freeze()
>>> di.health().healthy
False

ahealth async

ahealth() -> HealthReport

Run every declared check inside an event loop; the counterpart to health().

Drives async providers and async def checks. Otherwise identical.

Raises:

Type	Description
OutsideScopeError	A check's provider is scoped and no scope is active.
CircularDependencyError	This task re-enters construction of the same cached provider.

inject

inject(
    fn: Callable[P, Awaitable[R]],
) -> Callable[P, Awaitable[R]]

inject(fn: Callable[P, R]) -> Callable[P, R]

inject(fn: Callable[..., object]) -> Callable[..., object]

Wrap a function so parameters defaulting to injected are filled.

Returns a wrapper that, on each call, resolves every parameter whose default is injected and leaves the rest to the caller. The key comes from the parameter's annotation, in the same grammar a provider's parameters use: a class, Annotated[T, Tag(...)], Annotated[T, Named(...)], or T | None for a dependency that may be absent, which is filled with None when nothing provides it. Already-supplied arguments are never overridden, so an injected parameter can still be passed explicitly (handy in tests). The wrapper preserves the sync/async nature of fn. Injected keys are validated at decoration time, not call time: decorating raises immediately if a marked key is unregistered. Because the marker sits in default position, injected parameters must follow non-default ones or be keyword-only.

Raises:

Type	Description
InvalidProviderError	A marked parameter carries no annotation, or one whose names do not resolve.
MissingProviderError	A parameter requests an unregistered key and its annotation does not admit None.

Example

>>> from depin import Container, injected
>>> class Repo:
...     def count(self) -> int:
...         return 3
>>> di = Container().bind(Repo).freeze()
>>> @di.inject
... def handler(label: str, repo: Repo = injected) -> str:
...     return f'{label}={repo.count()}'
>>> handler(label='n')
'n=3'

override

override(
    key: ProviderKey, /, *, tag: str | None = None
) -> ProviderOverride

Select a provider to temporarily replace with ProviderOverride.using().

using() supplies a replacement for one block. Every resolution of the selected key and tag returns that replacement during the block, including resolutions deep in the graph. The override is bound to the current contextvars.Context, so concurrent contexts are unaffected; overrides nest and the innermost one wins.

Raises:

Type	Description
MissingProviderError	key is not a valid provider key type.

Example

>>> from depin import Container
>>> class Clock:
...     def now(self) -> str:
...         return 'real'
>>> class FakeClock:
...     def now(self) -> str:
...         return 'fake'
>>> di = Container().bind(Clock).freeze()
>>> with di.override(Clock).using(FakeClock()):
...     di[Clock].now()
'fake'
>>> di[Clock].now()
'real'

graph

graph() -> DependencyGraph

Return the validated dependency graph as data.

The view describes the plan Container.freeze() validated. An active override() does not change it, so the graph and both of its exports are the same on every call and in every context.

Example

>>> from depin import Container
>>> class Config: ...
>>> class Service:
...     def __init__(self, config: Config) -> None: ...
>>> di = Container().bind(Config).bind(Service).freeze()
>>> len(di.graph().nodes)
2
>>> di.graph().node(Service).dependencies[0].parameter
'config'

explain

explain(key: ProviderKey, *, tag: str | None = None) -> str

Return the resolution tree below a key, as text.

Each line names the parameter that requires the node, the node's key, its scope and provider shape, async when the node needs asynchronous resolution, and its tag when it has one. A subtree already shown is marked rather than repeated. A parameter that nothing provides is marked (unbound, default) when it carries a default, or (unbound, optional) when it does not but admits None.

A key no binding provides returns the line MissingProviderError carries for it, including the resolution chain when some provider requires that key. Like graph(), the output describes the validated plan, not an active override().

When the key is registered behind a condition that did not hold, the line says so, in the same wording Container.freeze() uses.

Raises:

Type	Description
MissingProviderError	The value cannot be a provider key at all. An unregistered key of a valid type is described in the returned text instead.

Example

>>> from depin import Container
>>> class Config: ...
>>> class Service:
...     def __init__(self, config: Config) -> None: ...
>>> di = Container().bind(Config).bind(Service).freeze()
>>> print(di.explain(Service))
Service  [singleton, class]
  config: Config  [singleton, class]