/litestar-queues

name: litestar-queues
description: "Use when working with litestar_queues, QueuePlugin, QueueConfig, WorkerConfig, QueueService, @task, QueuedBackgroundTask, QueueEventsConfig, or SQLSpecBackendConfig. Not for litestar-saq, Celery, or Dramatiq."

litestar-queues

`litestar-queues` is the first-party Litestar worker abstraction for task registration, durable queue state, worker lifecycle, schedules, uniqueness, bounded maintenance, execution dispatch, and application-facing task events.

Keep persistence and placement separate:

• A **queue backend** stores task records, identities, maintenance coordination, and optional event history.
• An **execution backend** decides where a claimed task runs.
• **Worker wakeups** are delivery hints. Persisted queue records remain the source of truth.

Code Style Rules

• Use `QueuePlugin` to wire lifecycle, application state, DI, task discovery, schedules, workers, and CLI commands.
• Put worker settings under `QueueConfig(worker=WorkerConfig(...))`.
• Inject `QueueService` with `NamedDependency[QueueService]`; never use a module-level service from handlers.
• Import public core types from `litestar_queues`. Import optional backend configuration from its backend submodule (`.sqlspec`, `.advanced_alchemy`, `.redis`, `.valkey`) and execution configs from `litestar_queues` or their respective submodules.
• Keep persistent-backend arguments and metadata JSON-serializable. Pass stable object IDs instead of large payloads.
• Use PEP 604 unions (`T | None`) and async I/O. Prefer `msgspec` for event/client DTOs unless the project already uses Pydantic.

Quick Reference

Minimal Plugin Setup

from litestar import Litestar, post
from litestar.di import NamedDependency
from litestar_queues import QueueConfig, QueuePlugin, QueueService, WorkerConfig, task


@task("accounts.sync", queue="accounts", retries=3, timeout=300)
async def sync_account(account_id: str) -> dict[str, str]:
    return {"account_id": account_id, "status": "synced"}


@post("/accounts/{account_id:str}/sync")
async def create_sync_job(
    account_id: str,
    queue_service: NamedDependency[QueueService],
) -> dict[str, str]:
    result = await queue_service.enqueue(sync_account, account_id)
    return {"task_id": str(result.id), "status": result.status or "queued"}


app = Litestar(
    route_handlers=[create_sync_job],
    plugins=[
        QueuePlugin(
            QueueConfig(worker=WorkerConfig(placement="server")),
        ),
    ],
)

`QueueConfig()` with no arguments defaults to `queue_backend="ephemeral"`, `execution_backend="local"`, and `placement="server"` — a private per-invocation SQLite database plus one CLI-owned worker process, with no broker, port, or extra dependency. Keep that shape for tests, development, and small single-process deployments only.

Process-local `"memory"` storage must be asked for explicitly, together with a placement that shares the process:

from litestar_queues import QueueConfig, WorkerConfig

config = QueueConfig(queue_backend="memory", worker=WorkerConfig(placement="asgi"))

Storage, execution, and placement combinations that cannot work are rejected at startup with a message naming the fix, rather than failing at first claim.

Task Options, Scheduling, and Uniqueness

from datetime import timedelta

from litestar_queues import QueueEventActor, QueueService, RetryBackoff, non_retryable, task


@task(
    "reports.render",
    queue="reports",
    priority=10,
    retries=3,
    retry_backoff=RetryBackoff(initial_delay=2.0, multiplier=2.0, max_delay=60.0),
    timeout=120,
    run_after=30,
    expires_in=timedelta(minutes=30),
    unique_by="arguments",
    unique_until="terminal",
    actor=QueueEventActor(type="service", id="reports-worker", name="Reports Worker"),
)
async def render_report(report_id: str, *, format: str = "pdf") -> str:
    if report_id == "invalid":
        non_retryable("Report ID does not exist")
    return f"{report_id}.{format}"


@task("reports.refresh", interval=timedelta(minutes=15), jitter=30, retries=2)
async def refresh_reports() -> None:
    pass


async def queue_report(queue_service: QueueService, report_id: str) -> str:
    result = await queue_service.enqueue(
        render_report,
        report_id,
        timeout=600,
        metadata={"requested_by": "system"},
    )
    await result.wait(timeout=30)
    return result.status or "unknown"

In addition to `await queue_service.enqueue(render_report, ...)` (the primary DI pattern in route handlers), decorated tasks expose `await render_report.enqueue(report_id)` and `await render_report.using(priority=20, run_after=60).enqueue(report_id)`. `Task.enqueue()` uses the active default `QueueService` registered by `QueuePlugin`, or falls back to an inline `"memory"` + `"immediate"` service when called in unit tests with no application wired. `result.record` holds the `QueuedTaskRecord` returned at enqueue time.

When enqueueing against an external execution backend such as Cloud Tasks, catch `QueueDispatchError` (`from litestar_queues.exceptions import QueueDispatchError`) and inspect `exc.committed`: when `True`, the record (`exc.task_id`) is already durably committed in SQL and bounded maintenance will repair the lost delivery.

Identity precedence is strict:

1. Explicit enqueue `key`. 2. Configured task `key`. 3. `unique_by="task"`. 4. `unique_by="arguments"`. 5. No identity.

`unique_until="terminal"` is the default and releases the identity after completion, failure, or cancellation. `unique_until="forever"` stores a permanent reservation until `await queue_service.reset_task_identity(effective_key)` removes it.

Do not combine a configured `key` with `unique_by`. Do not set `unique_until="forever"` without a configured `key` or `unique_by`. Use `QueueConfig.max_argument_identity_bytes` to bound canonical payloads hashed by `unique_by="arguments"`.

Use `interval` or five-field `cron`, never both. Re