Taskiq integration

MessageScope

Bases: TaskiqMiddleware

Taskiq middleware that opens one depin async scope around every message.

The scope opens before the task body runs and closes after it finishes, including when it finishes by raising: Taskiq calls post_execute for a failed task as well as a successful one, and the failure surfaces on the taskiq.TaskiqResult either way. Inside the scope the container is published to the message's context, so depin.hosted_container() reaches it from anywhere the task calls into, and the message is seeded under taskiq.TaskiqMessage. The scope's teardowns run when the message finishes, and the publication is undone after them.

The seeding is half of a pair the container has to complete. A provider that declares a taskiq.TaskiqMessage parameter — and is then handed the task id, name, labels and arguments of the message it is resolving for — resolves only if the container declared that key with depin.Container.scope_value; without it freeze() reports depin.errors.MissingProviderError for taskiq.TaskiqMessage. What the middleware supplies is the value, never the binding.

The scope is asynchronous, so an async task body has a place to await depin.FrozenContainer.aresolve and an async provider a place to run. A synchronous task body reaches the container too — from taskiq 0.11.19, the release whose receiver runs a def body under contextvars.copy_context, so the executor thread carries the message's context — but it cannot await, so it is limited to depin.FrozenContainer.resolve, and an async provider asked for there raises depin.errors.AsyncInSyncContextError.

Register it last. Taskiq runs pre_execute in registration order and post_execute in reverse, so a middleware registered after this one that raises in its own pre_execute skips this one's post_execute: the scope stays open and its teardowns never run.

There is no counterpart that spans the broker's lifetime. Taskiq's startup and shutdown hooks run in different asyncio tasks, so a publication opened in one cannot be undone in the other — depin.Host.activated() raises ValueError: Token was created in a different Context. Singletons need no such counterpart: they are cached on the container itself, so every message shares them.

Parameters:

Name	Type	Description	Default
container	FrozenContainer	The frozen container to host for the duration of each message.	required

Raises:

Type	Description
ExceptionGroup	One or more teardowns failed when the message's scope closed. Every failure is included; one does not hide another.

Example

Declare the message as a scoped value, then register the middleware on the broker after every other one::

di = Container().scope_value(TaskiqMessage).bind(Report, scope=Scope.SCOPED).freeze()

broker = InMemoryBroker().with_middlewares(SimpleRetryMiddleware(), MessageScope(di))

Report then declares a TaskiqMessage parameter and is built once per message, against the message being executed.

pre_execute async

pre_execute(message: TaskiqMessage) -> TaskiqMessage

Open the message's scope, seed the message into it, and hand the message on.

Parameters:

Name	Type	Description	Default
message	TaskiqMessage	The message about to be executed.	required

Returns:

Type	Description
TaskiqMessage	The message it was given, unmodified. Taskiq threads the return
TaskiqMessage	value of each pre_execute into the next one and finally into the
TaskiqMessage	task body, so returning anything else replaces the message.

post_execute async

post_execute(
    message: TaskiqMessage, result: TaskiqResult[object]
) -> None

Close the message's scope, running its teardowns and undoing the publication.

Closes the innermost scope still open in this context, and does nothing when none is — which is what a message whose pre_execute never ran looks like.

Parameters:

Name	Type	Description	Default
message	TaskiqMessage	The message that finished.	required
result	TaskiqResult[object]	The outcome Taskiq recorded for it, successful or failed.	required