Task Dependency Injection¶

Background tasks execute outside the scope of an incoming HTTP request. litestar-queues provides two complementary hooks on QueueConfig for injecting database sessions, API clients, and domain services into task executions:

• task_dependency_provider: Attempt-scoped async context manager that sets up and tears down resources around each task execution attempt.
• task_dependency_resolver: Stateless async callable that returns a dictionary of extra keyword arguments before the task body runs.

Runnable Example¶

"""Supplying attempt-scoped resources and stateless dependencies to tasks."""

from collections.abc import AsyncIterator, Mapping
from contextlib import asynccontextmanager

from litestar_queues import QueueConfig, QueueService, Task, TaskExecutionContext, WorkerConfig, task
from litestar_queues.models import QueuedTaskRecord


class BillingClient:
    """Example scoped client managed per task execution attempt."""

    __slots__ = ("closed", "prefix")

    def __init__(self, prefix: str) -> None:
        self.prefix = prefix
        self.closed = False

    async def charge(self, invoice_id: str) -> str:
        """Issue a charge reference."""
        return f"{self.prefix}:{invoice_id}"


@task("billing.capture_payment", queue="payments")
async def capture_payment(invoice_id: str, billing_client: BillingClient) -> str:
    """Execute a task receiving only the parameters declared in its signature."""
    return await billing_client.charge(invoice_id)


@asynccontextmanager
async def provide_task_dependencies(
    task: Task[..., object], record: QueuedTaskRecord, context: TaskExecutionContext
) -> AsyncIterator[Mapping[str, object]]:
    """Yield attempt-scoped dependencies and guarantee teardown on exit."""
    _ = (task, context)
    client = BillingClient(prefix=record.queue)
    try:
        yield {"billing_client": client, "unused_dependency": "filtered-out"}
    finally:
        client.closed = True


async def main() -> None:
    """Verify attempt-scoped dependency injection and automatic signature filtering."""
    config = QueueConfig(
        queue_backend="memory",
        execution_backend="immediate",
        worker=WorkerConfig(placement="external"),
        task_dependency_provider=provide_task_dependencies,
    )
    async with QueueService(config=config) as service:
        handle = await service.enqueue(capture_payment, "inv-500")
        await handle.refresh()
        assert handle.status == "completed"
        assert handle.result == "payments:inv-500"

Attempt-Scoped Providers vs Stateless Resolvers¶

Automatic Signature Filtering¶

Both task_dependency_provider and task_dependency_resolver can yield a shared mapping of application dependencies without coupling every task signature to the full dictionary:

1. Standard Signatures: When a task function does not declare a **kwargs (VAR_KEYWORD) parameter, litestar-queues filters the yielded mapping to inject only keys matching parameters declared in the task's inspect.Signature that were not already supplied explicitly at enqueue() time.
2. Variadic Signatures (**kwargs): When a task declares **kwargs, all non-overlapping keys from the provider or resolver mapping are forwarded.
3. Explicit Enqueue Precedence: Arguments passed explicitly to await queue_service.enqueue(task_fn, ...) always take precedence over keys yielded by a provider or resolver.

Lifecycle & Teardown Guarantees¶

When task_dependency_provider is configured, QueueService enforces strict resource safety:

• Guaranteed Exit: Once __aenter__ succeeds, __aexit__ is awaited exactly once for every attempt outcome: completion, retryable exception, non_retryable() failure, job_cancelled(), attempt timeout, external cancellation, heartbeat claim loss, and worker shutdown interruption.
• Coroutine Cancellation Semantics: In __aexit__(exc_type, exc_val, exc_tb), exc_type receives asyncio.CancelledError for attempt timeout, external cancellation, heartbeat claim loss, and worker shutdown alike because all four interrupt the running coroutine via cancellation. Inspect record and context (for example context.is_cancelled) rather than branching on exc_type alone.
• No Exception Suppression: Returning a truthy value from __aexit__ is ignored by QueueService; the provider is strictly a resource lifecycle scope, never an exception filter.
• Process-Level open() / close(): If the provider object defines open() and/or close() methods (sync or async), QueueService.open() invokes provider.open() before opening queue backends and QueueService.close() invokes provider.close() during shutdown (including partial-open rollback).