litestar-org/litestar-skills/litestar-queues

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. Recurring tasks inherit configured retries on startup and succession, and persisted occurrences retain their retry budget and backoff until completion. Use task_modules=("app.domain.reports.jobs",), discover_tasks("app.domain", "jobs"), or AutowirePlugin(AutowireConfig(domain_packages=["app.domain"], integrations=["dishka", "queues"])) before string enqueueing or schedule initialization.

To mark a failure permanent and bypass retries, call non_retryable("message") or raise NonRetryableError. To cooperatively cancel execution from within a handler, call job_cancelled("message") or raise JobCancelledError. Calling await queue_service.cancel_task(task_id, include_running=True) cancels any active remote provider execution (Cloud Run Jobs or Cloud Tasks) first before writing durable cancellation state.

Dependency Injection, Error Sanitization, and Role-Aware Worker Factory

• Attempt-scoped DI (`task_dependency_provider`): Enters an @asynccontextmanager per attempt (task, record, context) -> AsyncIterator[Mapping[str, Any]], merges the yielded mapping into task keyword arguments, and guarantees clean exit on success, error, timeout, or cancellation. When using Dishka, reuse an active in-process request container (request_container_var.get()) when running inline or open a fresh Scope.REQUEST scope on the worker container.
• Error sanitization & stale priority: Set error_sanitizer=Callable[[BaseException, QueuedTaskRecord], str] to convert exceptions into persisted error strings, and stale_requeue_priority ("preserve", int, or Callable[[int], int]) for stale task recovery.
• Worker placement (`WorkerConfig.placement`):
• "server" (default): starts one worker in the CLI server lifespan (litestar run).
• "asgi": starts one worker per ASGI worker in the application lifespan (e.g., target == "local" and role == "web").
• "external": starts nothing automatically; managed processes run litestar queues run or litestar queues run-consumer (or receive push deliveries on role == "consumer").
• Role-aware factory (`create_queue_config`): Mount EventStreamConfig only on the "web" role (None on "consumer" and "cli"), and harden CloudTasksExecutionConfig by enabling trust_platform_auth=(role == "consumer") and attaching a deny guard when role != "consumer". See Backends, Worker Configuration, Events, and Maintenance for complete factory and Dishka container examples.

LITESTAR_APP=app:app litestar queues run --queue default --max-concurrency 4 --drain-timeout 60
LITESTAR_APP=app:app litestar queues run-consumer --backend sqs --max-concurrency 4 --drain-timeout 60
LITESTAR_APP=app:app litestar queues status --json
LITESTAR_APP=app:app litestar queues scheduler-health --minutes 5
LITESTAR_APP=app:app litestar queues run-maintenance --json

Backend Selection

Queue backends own persistence:

Execution backends decide where a claimed task runs, and never own queue state:

Match the project's existing data stack. Memory cannot coordinate separate processes and has no MemoryBackendConfig class — pass queue_backend="memory" or an InMemoryQueueBackend instance. See Execution Backends and Backends, Worker Configuration, Events, and Maintenance.

Persistent Backends & Schema

• SQLSpec (`SQLSpecBackendConfig`): When manage_schema=True (default), QueuePlugin registers the 0001_create_queue_tasks migration revision. When manage_schema=False, register "litestar_queues" in migration_config={"include_extensions": ["events", "litestar_queues"]} and wire configure_queue_migration_extension(...) from litestar_queues.backends.sqlspec.extension. worker_wakeups defaults to SQLSpecWorkerWakeupConfig (notify_queue on PostgreSQL, poll_queue on DuckDB); set worker_wakeups=None with CloudTasksExecutionConfig.
• Advanced Alchemy (`SQLAlchemyBackendConfig`): Pass sqlalchemy_config=alchemy_config and manage schema via application Alembic migrations (QueueTaskModelMixin, QueueEventHistoryModelMixin, QueueMaintenanceModelMixin, QueueTaskReservationModelMixin). Set worker_wakeups=True only on PostgreSQL.
• Redis & Valkey (`RedisBackendConfig` / `ValkeyBackendConfig`): Use pub/sub wakeup hints by default with keys namespaced under key_prefix.