Integration contract¶
The seam an integration uses to host a container inside a framework. See Writing an integration for the narrative.
Host ¶
A FrozenContainer hosted inside a framework.
An integration builds one Host when the application is wired, then opens
a scope per unit of work — an HTTP request, a CLI invocation, a queue
message. Inside that scope the container is published to the current
contextvars.Context, so code that carries only an annotation reaches it
through hosted_container().
The publication is context-local, so concurrent requests and concurrent tasks never see each other's container, and two hosts in one process nest the published container: the innermost wins and the enclosing one is restored on exit.
Scopes do not nest that way. The scope frame stack is process-wide and shared by every container, so a scope opened inside another host's scope becomes a child of that frame, and a key already cached there is what the inner scope resolves — the inner container's own binding never runs. Two different containers must not nest their scopes.
Example
>>> from depin import Container, Host, Token, hosted_container
>>> request_id = Token[str]('request_id')
>>> di = Container().scope_value(request_id).freeze()
>>> host = Host(di)
>>> with host.scope() as frame:
... frame.provide(request_id, 'r-1')
... hosted_container().resolve(request_id)
'r-1'
activated ¶
activated() -> Generator[None]
Publish the container for the duration of the block, opening no scope.
What an integration uses outside a unit of work: an ASGI lifespan, a
process-wide CLI setup, anything that resolves singletons through
hosted_container() without a scope to open.
Example
>>> from depin import Container, Host, hosted_container
>>> di = Container().freeze()
>>> with Host(di).activated():
... hosted_container() is di
True
scope ¶
scope() -> Generator[ScopeFrame]
Publish the container and open one synchronous scope around a unit of work.
Yields the scope's frame so the caller can seed the framework's own
objects into it with ScopeFrame.provide before anything resolves. On
exit the scope's teardowns run first and the publication is undone
after, so a teardown can still reach the container.
Raises:
| Type | Description |
|---|---|
TeardownError
|
An async provider left a teardown in this sync
scope. Use |
ExceptionGroup
|
One or more teardowns failed when the scope closed. Every failure is included; one does not hide another. |
Example
>>> from depin import Container, Host, Token, hosted_container
>>> job = Token[str]('job')
>>> di = Container().scope_value(job).freeze()
>>> with Host(di).scope() as frame:
... frame.provide(job, 'reindex')
... hosted_container().resolve(job)
'reindex'
ascope
async
¶
ascope() -> AsyncGenerator[ScopeFrame]
Publish the container and open one asynchronous scope; the counterpart to scope().
Required when any provider in the scope is async. Otherwise identical: the frame is yielded for seeding, teardowns run before the publication is undone.
Raises:
| Type | Description |
|---|---|
ExceptionGroup
|
One or more teardowns failed when the scope closed. Every failure is included; one does not hide another. |
Example
>>> import asyncio
>>> from depin import Container, Host, Token, hosted_container
>>> job = Token[str]('job')
>>> di = Container().scope_value(job).freeze()
>>> async def run() -> None:
... async with Host(di).ascope() as frame:
... frame.provide(job, 'reindex')
... print(await hosted_container().aresolve(job))
>>> asyncio.run(run())
reindex
hosted_container ¶
hosted_container() -> FrozenContainer
Return the container hosted in this context.
Raises:
| Type | Description |
|---|---|
ContainerNotBoundError
|
No |
Example
>>> from depin import Container, Host, hosted_container
>>> di = Container().freeze()
>>> with Host(di).activated():
... hosted_container() is di
True
optional_hosted_container ¶
optional_hosted_container() -> FrozenContainer | None
Return the container hosted in this context, or None when there is none.
The non-raising counterpart to hosted_container(). An integration uses it
to raise its own message, naming the setup step its users actually
forgot — installing a middleware, registering a plugin — rather than the
contract-level one.
Example
>>> from depin import optional_hosted_container
>>> optional_hosted_container() is None
True
ContractVersion
dataclass
¶
The version of the integration contract a release of depin implements.
The minor number rises when an operation is added and every existing one
keeps its meaning; the major number rises when an operation changes meaning
or is removed. An integration that needs an operation added in 1.2
guards on depin.CONTRACT_VERSION >= ContractVersion(1, 2).
Attributes:
| Name | Type | Description |
|---|---|---|
major |
int
|
Rises on a breaking change to an existing operation. |
minor |
int
|
Rises when an operation is added. |
Example
>>> from depin import CONTRACT_VERSION, ContractVersion
>>> CONTRACT_VERSION >= ContractVersion(1, 0)
True
>>> str(CONTRACT_VERSION)
'1.0'
CONTRACT_VERSION
module-attribute
¶
CONTRACT_VERSION: Final = ContractVersion(1, 0)
The contract version this release of depin implements.