Litestar Queues

litestar-queues

litestar-queues 0.9.0 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 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",
)
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)
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"

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. Use task_modules=("app.tasks",) or discover_tasks("app.domain") 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.

Dependency Injection and Error Sanitization

Supply attempt-scoped dependencies to task handlers through task_dependency_resolver or task_dependency_provider:

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

from litestar_queues import QueueConfig, Task, TaskExecutionContext, QueuedTaskRecord


@asynccontextmanager
async def provide_task_dependencies(
    task: Task[Any, Any],
    record: QueuedTaskRecord,
    context: TaskExecutionContext,
) -> AsyncIterator[Mapping[str, Any]]:
    yield {"attempt_id": str(record.id)}


def sanitize_task_error(exc: BaseException, record: QueuedTaskRecord) -> str:
    return "An internal error occurred during task processing"


queue_config = QueueConfig(
    queue_backend="sqlspec",
    task_dependency_provider=provide_task_dependencies,
    error_sanitizer=sanitize_task_error,
    stale_requeue_priority="preserve",
)

task_dependency_provider enters an async context manager for each attempt, merges the yielded mapping into task keyword arguments, and guarantees clean exit on success, error, timeout, or cancellation.

error_sanitizer converts a raw exception into the task's persisted error string.

stale_requeue_priority controls priority on stale task recovery ("preserve", an int, or a Callable[[int], int]).

Worker Configuration and CLI

from litestar_queues import QueueConfig, WorkerConfig

queue_config = QueueConfig(
    queue_backend="sqlspec",
    worker=WorkerConfig(
        placement="external",
        batch_size=10,
        max_concurrency=8,
        queue_concurrency={"high_priority": 4, "default": 4},
        queues=("high_priority", "default"),
        poll_interval=0.25,
        heartbeat_interval=30,
        heartbeat_miss_threshold=2,
        graceful_shutdown_timeout=60,
        requeue_on_shutdown=False,
    ),
)

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

WorkerConfig.placement decides which process owns the worker:

• "server" (the default): exactly one worker in the CLI server lifespan (litestar run).
• "asgi": starts one worker per ASGI worker in the application lifespan.
• "external": starts nothing automatically; managed processes run litestar queues run or litestar queues run-consumer.

The worker adaptively backs off empty polling from poll_interval toward poll_backoff_max, using poll_backoff_multiplier and poll_jitter. Backend notifications can end the wait early; they never replace polling or durable state checks.

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. See Execution Backends for the managed transports.

SQLSpec Backend

from sqlspec.adapters.aiosqlite import AiosqliteConfig

from litestar_queues import QueueConfig
from litestar_queues.backends.sqlspec import SQLSpecBackendConfig

sqlspec_config = AiosqliteConfig(
    connection_config={"database": "queue.db"},
)

queue_config = QueueConfig(
    queue_backend=SQLSpecBackendConfig(
        sqlspec_config=sqlspec_config,
        manage_schema=True,
    ),
    execution_backend="local",
)

QueuePlugin registers the package migration with the supplied SQLSpec configuration. Run it through the application's normal SQLSpec migration workflow; opening the backend does not migrate the database. Use explicit create_schema() only for local bootstrap.

SQLSpecBackendConfig.worker_wakeups defaults to native wakeups. Capable PostgreSQL adapters use notify_queue, DuckDB uses poll_queue, and other adapters fall back to polling.

Advanced Alchemy Backend

from advanced_alchemy.extensions.litestar import SQLAlchemyAsyncConfig

from litestar_queues import QueueConfig
from litestar_queues.backends.advanced_alchemy import SQLAlchemyBackendConfig

alchemy_config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///queue.db",
)

queue_config = QueueConfig(
    queue_backend=SQLAlchemyBackendConfig(
        sqlalchemy_config=alchemy_config,
        worker_wakeups=False,
    ),
    execution_backend="local",
)

Use the application's Advanced Alchemy metadata and migration lifecycle. Compose the package mixins into adopter-owned models when custom bases, table names, or binds are required, then pass all matching model classes:

• model_class
• event_history_model_class
• maintenance_model_class
• task_reservation_model_class

Set worker_wakeups=True only for a supported PostgreSQL dialect.

Redis and Valkey Backends

from litestar_queues import QueueConfig
from litestar_queues.backends.redis import RedisBackendConfig

queue_config = QueueConfig(
    queue_backend=RedisBackendConfig(
        url="redis://localhost:6379/0",
        key_prefix="litestar_queues",
        worker_wakeups=True,
    ),
    execution_backend="local",
)

Use ValkeyBackendConfig from litestar_queues.backends.valkey for Valkey. Both use pub/sub wakeup hints by default. Their queue data, wakeup channel, maintenance key, and permanent reservations remain namespaced by the configured prefix.

Persistent Schema

SQL-backed deployments use queue records, maintenance coordination, and forever-uniqueness reservations. Add durable event history only when QueueEventsConfig.history is configured:

SQLSpec's packaged 0001_create_queue_tasks migration provisions the enabled tables and supports explicit table-name overrides. Advanced Alchemy applications own equivalent models and Alembic migrations. Redis and Valkey use namespaced keys and require no SQL migration.

Do not delete the reservation table during ordinary task or event retention. Forever identities are removed only through QueueService.reset_task_identity().

Events, Logging, and Progress

from litestar_queues import QueueConfig, task
from litestar_queues.events import (
    EventDeliveryConfig,
    QueueEventsConfig,
    beat,
    publish_task_event,
    publish_task_log,
    publish_task_progress,
)

queue_config = QueueConfig(
    events=QueueEventsConfig(
        channels=channels_backend,
        delivery=EventDeliveryConfig(
            publish_global_lifecycle=True,
        ),
    ),
)


@task("imports.process", timeout=300)
async def process_import(path: str) -> None:
    beat("validating")
    await publish_task_log("Import started", payload={"path": path})
    await publish_task_progress(current=50, total=100, message="Halfway")
    await publish_task_event("custom.checkpoint", message="Checkpoint reached")

QueueEventsConfig groups live delivery, application stream, and durable history. It must enable at least one capability. Task events are separate from worker wakeups.

Observability and Telemetry

from litestar_queues import QueueConfig
from litestar_queues.observability import ObservabilityConfig

queue_config = QueueConfig(
    queue_backend="sqlspec",
    observability=ObservabilityConfig(
        enable_otel=True,
        enable_prometheus=True,
        enable_sqlcommenter=True,
    ),
    scheduler_canary_task="health.canary",
)

ObservabilityConfig configures OpenTelemetry tracing and Prometheus metrics. OpenTelemetry trace contexts and correlation IDs propagate across dispatchers, broker execution backends, and workers.

The scheduler health command (litestar queues scheduler-health --minutes 5) verifies that the configured canary task completed within the specified window.

Bounded Maintenance

from litestar_queues import QueueConfig, QueueMaintenanceConfig

queue_config = QueueConfig(
    queue_backend="sqlspec",
    maintenance=QueueMaintenanceConfig(
        time_budget=300,
        coordination_timeout=360,
        stale_after=900,
        stale_limit=100,
        terminal_retention=30 * 24 * 60 * 60,
        terminal_limit=1000,
        event_limit=1000,
    ),
)

LITESTAR_APP=app:app litestar queues run-maintenance --json
LITESTAR_APP=app:app litestar queues run-maintenance --phase stale --phase terminal

One maintenance invocation runs bounded phases in fixed order: external reconciliation, stale recovery, terminal retention, then event retention. It never starts a worker, executes queued work, or loops to drain a backlog.

Schedule one external six-hour or daily invocation. Retention phases have no destructive defaults: stale_after, terminal_retention, and event rules remain disabled when None. coordination_timeout must exceed time_budget.

External One-Task Execution

LITESTAR_QUEUES_TASK_ID=4d821c46-8c60-4ec3-b884-3f62eb71a03e \
LITESTAR_QUEUES_CONFIG_FACTORY=app.queue:create_queue_config \
litestar queues run-task

run-task claims and executes one existing record for an external executor. --task-id, --config-factory, and --task-modules override its environment inputs for manual operation. It is not a standalone worker loop.

Background Responses

from litestar import Response, post
from litestar_queues import QueuedBackgroundTask, task


@task("imports.process")
async def process_import(path: str) -> None:
    pass


@post("/imports")
async def create_import() -> Response[dict[str, str]]:
    return Response(
        {"status": "accepted"},
        background=QueuedBackgroundTask(process_import, "/tmp/data.csv"),
    )

QueuedBackgroundTask enqueues after the response is sent. It resolves the active plugin service at construction; pass service=queue_service for a custom service.

Workflow

1. Identify the work shape. Use Litestar Queues for durable task state, worker placement, schedules, progress/events, task uniqueness, or bounded maintenance.
2. Match the queue backend. Use memory for same-process tests, SQLSpec for SQLSpec apps, Advanced Alchemy for SQLAlchemy apps, Redis for Redis infrastructure, or Valkey for Valkey infrastructure.
3. Pick execution placement. Use local workers by default, immediate execution for tests/scripts, broker consumers (Kafka, PubSub, RabbitMQ, SQS) for distributed broker dispatch, and Cloud Run for isolated external jobs.
4. Configure the worker. Put every worker setting under WorkerConfig; decide explicitly whether it runs in the app lifespan.
5. Provision persistent storage. Run SQLSpec migrations or add all required Advanced Alchemy models to application-owned migrations.
6. Define and discover tasks. Decorate callables with @task; import their modules before string enqueueing or schedule initialization.
7. Enqueue through DI. Inject QueueService, enqueue a decorated task or registered name, and wait only when the caller truly needs the terminal state.
8. Add optional capabilities separately. Configure worker wakeups, task-event delivery/history, permanent uniqueness, observability, and maintenance only when their storage and operational lifecycles are owned.
9. Place operational commands. Run standalone workers or continuous broker consumers (run-consumer) continuously, run-task only in external one-task executors, and maintenance from one infrequent external schedule.

Guardrails

• Use SQLAlchemyBackendConfig for Advanced Alchemy persistence. Import it from litestar_queues.backends.advanced_alchemy.
• Configure events with QueueEventsConfig. Add EventDeliveryConfig, EventStreamConfig, and/or EventHistoryConfig for the required capabilities.
• Do not pass flat worker fields to QueueConfig. Use QueueConfig(worker=WorkerConfig(...)).
• Do not use memory across processes. It cannot coordinate standalone workers or a separate maintenance command.
• Do not confuse wakeups with task events. Wakeups hint that work may exist; task events serve application and operator consumers.
• Do not rely on wakeups for correctness. Notifications may be delayed or lost; poll and claim durable state.
• Do not let the backend open path own migrations. Run SQLSpec migrations explicitly or manage Advanced Alchemy schema in the application.
• Do not omit maintenance or reservation storage. Provision all four SQL concerns used by the deployment.
• Do not enable retention implicitly. Leave each destructive threshold at None until an explicit policy exists.
• Do not run maintenance as a minute-level task. Use one bounded external invocation every six hours or daily.
• Do not combine key and unique_by. Select one identity source.
• Do not assume run-task starts a worker. It consumes one already-persisted record and exits.
• Do not import optional configs from litestar_queues.backends. Use the concrete .sqlspec, .advanced_alchemy, .redis, or .valkey module.
• Do not enqueue by string before task discovery. Import task modules or call discover_tasks().

Validation Checkpoint

• The dependency floor is litestar-queues>=0.9.0
• QueuePlugin receives one QueueConfig
• Worker options live under QueueConfig.worker
• WorkerConfig.placement matches the deployment topology
• Standalone workers use a shared persistent queue backend
• The backend matches the project's SQLSpec, Advanced Alchemy, Redis, or Valkey stack
• SQLSpec uses sqlspec_config and the normal migration workflow
• Advanced Alchemy uses SQLAlchemyBackendConfig and application-owned migrations
• All enabled SQL concerns have queue, event-history, maintenance, and reservation tables
• Worker wakeups are configured independently from application task events
• QueueEventsConfig enables delivery, stream, or history
• Task modules are loaded before string enqueueing and schedule initialization
• Handlers inject NamedDependency[QueueService]
• Uniqueness uses one identity source and the intended lifetime
• max_argument_identity_bytes bounds untrusted argument-derived identity inputs
• Maintenance thresholds and one external schedule are explicit
• run, run-consumer, run-task, and run-maintenance are used for their distinct lifecycles

Example

Task: Persist report jobs with SQLSpec, run a standalone worker, deduplicate equivalent calls, and perform bounded retention.

from litestar import Litestar, post
from litestar.di import NamedDependency
from sqlspec.adapters.aiosqlite import AiosqliteConfig

from litestar_queues import (
    QueueConfig,
    QueueMaintenanceConfig,
    QueuePlugin,
    QueueService,
    WorkerConfig,
    task,
)
from litestar_queues.backends.sqlspec import SQLSpecBackendConfig


@task(
    "reports.render",
    queue="reports",
    retries=2,
    timeout=300,
    unique_by="arguments",
)
async def render_report(report_id: str) -> dict[str, str]:
    return {"report_id": report_id, "status": "rendered"}


@post("/reports/{report_id:str}/render")
async def enqueue_report(
    report_id: str,
    queue_service: NamedDependency[QueueService],
) -> dict[str, str]:
    result = await queue_service.enqueue(render_report, report_id)
    return {"task_id": str(result.id), "status": result.status or "queued"}


sqlspec_config = AiosqliteConfig(
    connection_config={"database": "queue.db"},
)
queue_config = QueueConfig(
    queue_backend=SQLSpecBackendConfig(sqlspec_config=sqlspec_config),
    execution_backend="local",
    worker=WorkerConfig(
        placement="external",
        max_concurrency=4,
        queues=("reports",),
    ),
    maintenance=QueueMaintenanceConfig(
        time_budget=300,
        coordination_timeout=360,
        stale_after=900,
        terminal_retention=30 * 24 * 60 * 60,
    ),
    task_modules=("app.tasks",),
    max_argument_identity_bytes=64 * 1024,
)

app = Litestar(
    route_handlers=[enqueue_report],
    plugins=[QueuePlugin(queue_config)],
)

LITESTAR_APP=app:app litestar queues run
LITESTAR_APP=app:app litestar queues run-maintenance --json

References Index

• Execution Backends — local, immediate, Cloud Run, Cloud Tasks, Kafka, Pub/Sub, RabbitMQ, SQS, delivery fencing, and lost-delivery repair.
• Namespacing and Expiry — namespace=, not-started deadlines and the expired state, event-stream configuration.

Cross-References

• litestar — Litestar app setup, plugin lists, DI, and lifespan.
• litestar-routing — Route handlers and controllers for enqueue endpoints.
• litestar-di — NamedDependency and service injection.
• litestar-plugins — Plugin initialization and lifecycle.
• litestar-autowire — Optional task discovery through Autowire integration.
• litestar-realtime — Channels, SSE, WebSockets, and task-event fan-out.
• litestar-testing — Application and handler tests.
• sqlspec — SQLSpec adapter and migration configuration.
• advanced-alchemy — SQLAlchemy models, services, and Alembic ownership.

Official References

• https://github.com/cofin/litestar-queues/tree/v0.9.0
• https://github.com/cofin/litestar-queues/releases/tag/v0.9.0
• https://github.com/cofin/litestar-queues/blob/v0.9.0/src/litestar_queues/config.py
• https://github.com/cofin/litestar-queues/blob/v0.9.0/src/litestar_queues/task.py
• https://github.com/cofin/litestar-queues/blob/v0.9.0/src/litestar_queues/_cli.py
• https://github.com/cofin/litestar-queues/blob/v0.9.0/src/litestar_queues/backends/sqlspec/config.py
• https://github.com/cofin/litestar-queues/blob/v0.9.0/src/litestar_queues/backends/advanced_alchemy/config.py
• https://github.com/cofin/litestar-queues/blob/v0.9.0/src/litestar_queues/maintenance.py

Shared Styleguide Baseline

• Use shared styleguides for generic language/framework rules to reduce duplication in this skill.
• General Principles
• Python
• Litestar
• Keep this skill focused on litestar-queues workflows, backend selection, worker placement, uniqueness, maintenance, and task-event APIs.