Defining & Enqueuing Tasks¶

Tasks are Python functions decorated with @task and registered in the process-wide task registry. Both async def coroutines and synchronous def functions are supported.

Registering Tasks with the Decorator¶

Applying @task wraps a callable in a Task descriptor while preserving direct invocation (await generate_invoice("ord-1001", 4999)) for unit tests and local callers.

"""Defining tasks with @task and enqueuing work through QueueService."""

from litestar_queues import QueueConfig, QueueService, WorkerConfig, task


@task("emails.send_welcome", queue="emails")
async def send_welcome_email(user_id: int, email: str) -> dict[str, object]:
    """Send a welcome email and return delivery metadata."""
    return {"user_id": user_id, "email": email, "delivered": True}


async def main() -> None:
    """Enqueue a task by callable and by registered name."""
    config = QueueConfig(
        queue_backend="memory", execution_backend="immediate", worker=WorkerConfig(placement="external")
    )
    async with QueueService(config=config) as service:
        first = await service.enqueue(send_welcome_email, 42, email="ada@example.com")
        await first.refresh()
        assert first.status == "completed"
        assert first.result == {"user_id": 42, "email": "ada@example.com", "delivered": True}

        second = await service.enqueue("emails.send_welcome", 99, email="grace@example.com")
        await second.refresh()
        assert second.status == "completed"
        assert second.result == {"user_id": 99, "email": "grace@example.com", "delivered": True}

Task Naming & Logical Queues¶

• Explicit Name: Pass a string as the first argument to @task("billing.generate_invoice", queue="billing") to assign a stable, refactoring-safe identifier persisted in QueuedTaskRecord.task_name.
• Default Name: When called as @task or @task() without a name, the task registers under function.__name__.
• Logical Queue: The queue argument (default "default") determines which logical queue receives the record unless overridden at enqueue time.

Enqueuing Work via QueueService¶

QueueService is the primary runtime facade injected into Litestar route handlers or opened as an async context manager in scripts and tests.

Single Enqueue¶

await queue_service.enqueue(task_or_name, *args, **kwargs) persists a QueuedTaskRecord in the configured storage backend, dispatches or wakes the target execution transport, and returns a TaskResult handle.

You can pass either the @task-decorated object or its registered string name:

• By Task Object: await service.enqueue(generate_invoice, "ord-1001", 4999) validates and normalizes arguments against the callable signature before persisting.
• By String Name: await service.enqueue("billing.generate_invoice", "ord-1002", 8500, queue="priority-billing") looks up the task if registered locally, or enqueues by raw name when producer processes do not import worker task modules.

Bulk Enqueue¶

When submitting batches of tasks in a single transaction or pipeline round-trip, construct TaskRequest items and call await queue_service.enqueue_many(requests):

from litestar_queues import TaskRequest

records = await queue_service.enqueue_many(
    [
        TaskRequest(task_name="billing.generate_invoice", args=("ord-2001", 1500), queue="billing"),
        TaskRequest(task_name="billing.generate_invoice", args=("ord-2002", 3200), queue="billing"),
    ]
)

enqueue_many returns list[QueuedTaskRecord] in the exact order of the input sequence.

Task Module Auto-Discovery¶

Workers and standalone CLI processes must import the modules where @task functions are defined so the task registry is populated before claims arrive. Configure QueueConfig.task_modules with module or package dotted paths:

from litestar_queues import QueueConfig

config = QueueConfig(
    queue_backend="memory",
    task_modules=["my_app.tasks", "my_app.billing.jobs"],
)

When QueuePlugin initializes or discover_tasks("my_app.tasks") runs, litestar-queues recursively imports packages and modules so every @task definition registers deterministically.

Execution Outcome Semantics¶

When a worker or one-shot consumer executes a claimed record via QueueService.execute_record:

• A return value becomes the completed result persisted on QueuedTaskRecord.result, and the task transitions to completed.
• A raised exception follows the configured retry policy: retryable exceptions increment retry_count and reschedule the record while retry_count < max_retries, whereas exhausting retries or calling non_retryable() transitions the record to failed.