litestar-queues · diff
git:20261020.150bdbe to git:20261104.3c5d082
35 added, 507 removed. Audit A to A.
---
name: litestar-queues
- description: "Auto-activate for litestar_queues, QueuePlugin, QueueConfig, WorkerConfig, QueueService, @task, QueuedBackgroundTask, QueueEventsConfig, or SQLSpecBackendConfig. Not for litestar-saq, Celery, or Dramatiq."
+ 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` 0.12.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.
+ `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
```python
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:
```python
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
```python
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)`, `await render_report.using(priority=20, run_after=60).enqueue(report_id)`, and direct invocation via `await render_report(report_id)` (`Task.__call__`), which runs the wrapped callable inline (offloading synchronous callables to a thread unless `sync_to_thread=False`) so other tasks or services can reuse task bodies without a second queue hop. `result.record` holds the `QueuedTaskRecord` returned at enqueue time.
-
- - **Reserved enqueue keyword arguments:** `Task.enqueue(**kwargs)` forwards keyword arguments directly to `QueueService.enqueue(task, *args, scheduled_at=..., run_after=..., expires_in=..., expires_at=..., key=..., queue=..., priority=..., retries=..., timeout=..., execution_backend=..., execution_profile=..., description=..., log_level=..., log_success=..., requeue_on_stale=..., metadata=..., **kwargs)`. `@task` function parameters must never use those 16 reserved keyword names (`scheduled_at`, `run_after`, `expires_in`, `expires_at`, `key`, `queue`, `priority`, `retries`, `timeout`, `execution_backend`, `execution_profile`, `description`, `log_level`, `log_success`, `requeue_on_stale`, `metadata`) because they are consumed as enqueue options instead of reaching `record.kwargs`.
- - **`Task.enqueue()` default-service fallback:** `Task.enqueue()` resolves `get_default_service()` (set by `QueuePlugin` lifespan startup) and **silently falls back** to an inline `"memory"` + `"immediate"` service when none is active (such as in unit tests with no application wired). Code that requires durable persistence outside request DI must resolve `QueueService` explicitly (`app.plugins.get(QueuePlugin).config.get_service(app.state)` or `get_default_service()` with an explicit `None` guard) rather than relying on `Task.enqueue()`.
- - **Foreground CLI execution (`execution_backend="immediate"`):** CLI commands that trigger `@task` jobs should not maintain separate `--enqueue` vs direct-call code paths. Route them through `await queue_service.enqueue(task, ..., execution_backend="immediate")` so the queue backend persists, claims, executes inline in the foreground via `ImmediateExecutionBackend`, and records terminal state and events through a single unified task pipeline.
+ 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 (`TaskReservation`, inspectable via `await queue_service.get_task_identity(effective_key)`): a collision returns a `TaskResult` pointing to the existing reservation's `task_id` without enqueueing a duplicate record, and the reservation persists after terminal states until `await queue_service.reset_task_identity(effective_key)` removes it — making `unique_until="forever"` + explicit `key` an idempotent "schedule once" primitive without external CAS locks.
+ `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`. Durable cancellation is cooperative: `await queue_service.cancel_task(task_id, include_running=True) -> bool` (returning `True` only for the caller that wins the durable state transition; or bulk `await queue_service.cancel_tasks(..., include_running=True) -> int`) cancels any active remote provider execution (Cloud Run Jobs or Cloud Tasks) first before writing durable `"cancelled"` state and notifying the worker control channel. When the worker reconciles running tasks (`WorkerConfig.cancellation_poll_interval`, default `1.0s`, or immediately on control notification), it marks `TaskExecutionContext` cancelled (`context.is_cancelled` / `await context.wait_cancelled()` / `context.raise_if_cancelled()`) and cancels the local task coroutine; long-running task bodies should race `await get_current_task_context().wait_cancelled()` or call `context.raise_if_cancelled()` and clean up resources in `try...finally` blocks.
-
- ### Dependency Injection and Error Sanitization
-
- Tasks have no implicit access to `app` or `app.state`. Supply attempt-scoped dependencies to task handlers through `task_dependency_resolver` (`TaskDependencyResolver`) or `task_dependency_provider` (`TaskDependencyProvider`), which are **mutually exclusive** on `QueueConfig`:
-
- ```python
- from collections.abc import AsyncGenerator, Mapping
- from contextlib import asynccontextmanager
- from typing import Any
-
- from litestar_queues import QueueConfig, QueuedTaskRecord, Task, TaskExecutionContext
-
-
- async def resolve_task_dependencies(
- task: Task[Any, Any],
- record: QueuedTaskRecord,
- context: TaskExecutionContext,
- ) -> Mapping[str, object]:
- if "attempt_id" not in task.signature.parameters:
- return {}
- return {"attempt_id": str(record.id)}
-
-
- @asynccontextmanager
- async def provide_task_dependencies(
- task: Task[Any, Any],
- record: QueuedTaskRecord,
- context: TaskExecutionContext,
- ) -> AsyncGenerator[Mapping[str, Any], None]:
- if "attempt_id" not in task.signature.parameters:
- yield {}
- return
- 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",
- )
- ```
-
- Use `task_dependency_resolver` when supplying stateless keyword arguments, and `task_dependency_provider` when the attempt owns a scoped resource: `task_dependency_provider` enters an async context manager inside the attempt timeout, merges the yielded mapping directly into task keyword arguments via `kwargs.update(extra_kwargs)` (inspect `task.signature.parameters` so you only yield keys accepted by that task), and guarantees clean `__aexit__` on success, error, timeout, claim loss, shutdown interruption, or cancellation (provider objects may also expose optional `open()` and `close()` lifecycle hooks managed by `QueueService`).
-
- When the application uses **Dishka**, task handlers can either receive dependencies from `task_dependency_provider` or resolve an async `Scope.REQUEST` container that reuses an active in-process request container (`request_container_var.get()`) when running inline and opens a fresh worker scope when running in an external worker:
-
- ```python
- from collections.abc import AsyncIterator
- from contextlib import asynccontextmanager
-
- from dishka import AsyncContainer, Scope
-
- from app.config.ioc import get_worker_container, request_container_var
-
-
- @asynccontextmanager
- async def resolve_task_container() -> AsyncIterator[AsyncContainer]:
- active = request_container_var.get()
- if active is not None:
- yield active
- return
- container = get_worker_container()
- async with container(scope=Scope.REQUEST) as request_container:
- token = request_container_var.set(request_container)
- try:
- yield request_container
- finally:
- request_container_var.reset(token)
- ```
-
- `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, Role-Aware Factory, and CLI
-
- ```python
- from typing import Any, Literal
-
- from litestar.connection import ASGIConnection
- from litestar.exceptions import PermissionDeniedException
-
- from litestar_queues import (
- CloudTasksExecutionConfig,
- EventDeliveryConfig,
- EventHistoryConfig,
- EventStreamConfig,
- QueueConfig,
- QueueEventsConfig,
- WorkerConfig,
- )
- from litestar_queues.backends.sqlspec import SQLSpecBackendConfig, SQLSpecWorkerWakeupConfig
-
- RuntimeRole = Literal["web", "consumer", "cli"]
-
-
- def deny_cloud_tasks_delivery_guard(connection: ASGIConnection[Any, Any, Any, Any], _: Any) -> None:
- raise PermissionDeniedException("Cloud Tasks push delivery is disabled on this runtime role")
+ 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
- def create_queue_config(
- *,
- sqlspec_config: Any,
- channels_backend: Any,
- target: Literal["local", "cloudtasks"] = "local",
- role: RuntimeRole = "web",
- ) -> QueueConfig:
- is_consumer = role == "consumer"
- use_cloud_tasks = target == "cloudtasks"
- execution_backend: Literal["local"] | CloudTasksExecutionConfig = (
- CloudTasksExecutionConfig(
- project_id="my-gcp-project",
- location="us-central1",
- queue_id="app-tasks",
- service_url="https://consumer.example.com",
- service_account_email="tasks-invoker@my-gcp-project.iam.gserviceaccount.com",
- trust_platform_auth=is_consumer,
- guards=() if is_consumer else (deny_cloud_tasks_delivery_guard,),
- )
- if use_cloud_tasks
- else "local"
- )
- return QueueConfig(
- queue_backend=SQLSpecBackendConfig(
- sqlspec_config=sqlspec_config,
- queue_table_name="queue_task",
- event_history_table_name="queue_task_event_history",
- worker_wakeups=None if use_cloud_tasks else SQLSpecWorkerWakeupConfig(queue_table_name="event_queue"),
- manage_schema=False,
- ),
- execution_backend=execution_backend,
- worker=WorkerConfig(
- placement="asgi" if (target == "local" and role == "web") else "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,
- ),
- events=QueueEventsConfig(
- channels=channels_backend,
- delivery=EventDeliveryConfig(publish_task_channel=True, publish_global_lifecycle=True),
- stream=EventStreamConfig(path="/queues/events", transports={"sse", "websocket"}) if role == "web" else None,
- history=EventHistoryConfig(batch_size=1),
- ),
- )
- ```
+ - **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](references/backends-and-events.md) for complete factory and Dishka container examples.
```bash
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 (for example, when `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"`).
-
- Mount `EventStreamConfig` only on the `"web"` role (`None` on `"consumer"` and `"cli"`), and harden `CloudTasksExecutionConfig` by enabling `trust_platform_auth=is_consumer` and attaching a deny guard when `not is_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:
| Existing stack or need | Queue backend | Import |
| --- | --- | --- |
| Default; dev and single-invocation apps | `"ephemeral"` | Core package (stdlib `sqlite3`) |
| Process-local tests, shared-process placement only | `"memory"` (`InMemoryQueueBackend`) | Core package (`litestar_queues`) |
| SQLSpec-managed SQL persistence | `SQLSpecBackendConfig(...)` | `litestar_queues.backends.sqlspec` |
| Advanced Alchemy / SQLAlchemy models | `SQLAlchemyBackendConfig(...)` | `litestar_queues.backends.advanced_alchemy` |
| Existing Redis infrastructure | `RedisBackendConfig(...)` | `litestar_queues.backends.redis` |
| Existing Valkey infrastructure | `ValkeyBackendConfig(...)` | `litestar_queues.backends.valkey` |
Execution backends decide where a claimed task runs, and never own queue state:
| Need | Execution backend | Import |
| --- | --- | --- |
| Default in-process workers | `"local"` | Core package |
| Inline completion in tests/scripts | `"immediate"` | Core package |
| Isolated Google Cloud Run Jobs | `CloudRunExecutionConfig(...)` | `litestar_queues.execution.cloudrun` |
| Serverless delivery with no worker process | `CloudTasksExecutionConfig(...)` | `litestar_queues.execution.cloudtasks` |
| Apache Kafka continuous consumer fleet | `KafkaExecutionConfig(...)` | `litestar_queues.execution.kafka` |
| GCP Pub/Sub continuous consumer fleet | `PubSubExecutionConfig(...)` | `litestar_queues.execution.pubsub` |
| RabbitMQ continuous consumer fleet (Python 3.11+) | `RabbitMQExecutionConfig(...)` | `litestar_queues.execution.rabbitmq` |
| Amazon SQS continuous consumer fleet | `SqsExecutionConfig(...)` | `litestar_queues.execution.sqs` |
- 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](references/execution-backends.md) for the managed transports.
-
- ### SQLSpec Backend
-
- Choose between **migration-managed** (`manage_schema=False`) and **auto-managed** (`manage_schema=True`) SQLSpec schema strategies:
-
- ```python
- from sqlspec.adapters.asyncpg import AsyncpgConfig
-
- from litestar_queues import QueueConfig
- from litestar_queues.backends.sqlspec import SQLSpecBackendConfig, SQLSpecWorkerWakeupConfig
- from litestar_queues.backends.sqlspec.extension import configure_queue_migration_extension
-
- sqlspec_config = AsyncpgConfig(
- connection_config={"dsn": "postgresql://postgres:postgres@127.0.0.1:5432/app"},
- migration_config={
- "script_location": "migrations",
- "include_extensions": ["events", "litestar_queues"],
- },
- extension_config={
- "events": {"queue_table_name": "event_queue"},
- },
- )
- configure_queue_migration_extension(
- sqlspec_config,
- queue_table_name="queue_task",
- event_history_enabled=True,
- event_history_table_name="queue_task_event_history",
- )
-
- migration_managed_config = QueueConfig(
- queue_backend=SQLSpecBackendConfig(
- sqlspec_config=sqlspec_config,
- queue_table_name="queue_task",
- event_history_table_name="queue_task_event_history",
- worker_wakeups=SQLSpecWorkerWakeupConfig(queue_table_name="event_queue"),
- manage_schema=False,
- ),
- execution_backend="local",
- )
-
- auto_managed_config = QueueConfig(
- queue_backend=SQLSpecBackendConfig(
- sqlspec_config=AsyncpgConfig(
- connection_config={"dsn": "postgresql://postgres:postgres@127.0.0.1:5432/app"},
- ),
- manage_schema=True,
- ),
- execution_backend="local",
- )
- ```
-
- When `manage_schema=True` (the default), `QueuePlugin` registers the consolidated `0001_create_queue_tasks` migration revision (including `dispatch_checked_at`) with the supplied SQLSpec configuration. When `manage_schema=False` (when SQLSpec migrations are owned directly on `AsyncpgConfig`), register `"litestar_queues"` in `migration_config={"include_extensions": ["events", "litestar_queues"]}` and wire `configure_queue_migration_extension(...)` (alongside `queue_migration_directory()`) from `litestar_queues.backends.sqlspec.extension`. Opening the backend never migrates the database; use `create_schema()` only for local bootstrap.
-
- `SQLSpecBackendConfig.worker_wakeups` defaults to native wakeups (`SQLSpecWorkerWakeupConfig`). Capable PostgreSQL adapters use `notify_queue`, DuckDB uses `poll_queue`, and other adapters fall back to polling. Set `worker_wakeups=None` when using `CloudTasksExecutionConfig` (no polling worker exists). SQLSpec wakeup transports validate that their events queue table exists before first use and raise `QueueConfigurationError` when missing.
-
- ### Advanced Alchemy Backend
-
- ```python
- 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 from `litestar_queues.backends.advanced_alchemy` into adopter-owned models when custom bases, table names, or binds are required, then pass all matching model classes:
-
- - `model_class` (`QueueTaskModelMixin`, default `QueueTaskModel`)
- - `event_history_model_class` (`QueueEventHistoryModelMixin`, default `QueueEventHistoryModel`)
- - `maintenance_model_class` (`QueueMaintenanceModelMixin`, default `QueueMaintenanceModel`)
- - `task_reservation_model_class` (`QueueTaskReservationModelMixin`, default `QueueTaskReservationModel`)
-
- Set `worker_wakeups=True` only for a supported PostgreSQL dialect.
-
- ### Redis and Valkey Backends
-
- ```python
- 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.
+ 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](references/execution-backends.md) and [Backends, Worker Configuration, Events, and Maintenance](references/backends-and-events.md).
- ### Persistent Schema
+ ### Persistent Backends & Schema
- SQL-backed deployments use queue records, maintenance coordination, and
- forever-uniqueness reservations. Add durable event history only when
- `QueueEventsConfig.history` is configured:
+ - **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`.
| Concern | Default table | Override setting (`SQLSpecBackendConfig`) |
| --- | --- | --- |
| Queue records | `queue_task` | `queue_table_name` |
| Distributed maintenance coordination | `queue_maintenance` | `maintenance_table_name` |
| Forever-uniqueness reservations | `queue_task_reservation` | `task_reservation_table_name` |
| Durable event history (when enabled) | `queue_task_event_history` | `event_history_table_name` |
- SQLSpec's packaged `0001_create_queue_tasks` migration provisions the enabled tables and supports explicit table-name overrides and promoted event-history columns (`event_history_extra_columns`). 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, Progress, and Event History Queries
-
- ```python
- from typing import Any
-
- from litestar_queues import QueueConfig, get_current_task_context, task
- from litestar_queues.events import (
- EventDeliveryConfig,
- EventHistoryConfig,
- EventHistoryExtraColumn,
- EventStreamConfig,
- QueueEventActor,
- QueueEventEntityRef,
- QueueEventQuery,
- QueueEventsConfig,
- beat,
- create_event_producer,
- publish_task_event,
- )
-
- queue_config = QueueConfig(
- events=QueueEventsConfig(
- channels=channels_backend,
- delivery=EventDeliveryConfig(
- publish_task_channel=True,
- publish_global_lifecycle=True,
- ),
- stream=EventStreamConfig(
- path="/queues/events",
- transports={"sse", "websocket"},
- ),
- history=EventHistoryConfig(
- batch_size=1,
- strict=True,
- max_pending=2000,
- extra_columns=(EventHistoryExtraColumn("tenant_id", "metadata.tenant_id", indexed=True),),
- ),
- ),
- )
-
-
- async def emit_progress(
- current: int,
- total: int | None,
- *,
- message: str,
- stage: str | None = None,
- entity: QueueEventEntityRef | None = None,
- payload: dict[str, Any] | None = None,
- ) -> None:
- context = get_current_task_context()
- if context is not None:
- event_payload = dict(payload or {})
- if stage is not None:
- event_payload["stage"] = stage
- await context.progress(
- current=current,
- total=total,
- message=message,
- entity=entity,
- payload=event_payload,
- )
-
-
- async def emit_log(
- message: str,
- *,
- level: str = "info",
- stage: str | None = None,
- entity: QueueEventEntityRef | None = None,
- payload: dict[str, Any] | None = None,
- ) -> None:
- context = get_current_task_context()
- if context is not None:
- event_payload = dict(payload or {})
- if stage is not None:
- event_payload["stage"] = stage
- await context.log(
- message,
- level=level,
- entity=entity,
- payload=event_payload,
- )
-
-
- @task("imports.process", timeout=300)
- async def process_import(path: str, tenant_id: str) -> None:
- beat("validating")
- entity = QueueEventEntityRef(type="import", id=path, name=path)
- actor = QueueEventActor(type="tenant", id=tenant_id)
- await emit_log("Import started", stage="validating", entity=entity, payload={"path": path})
- await emit_progress(50, 100, message="Halfway", stage="importing", entity=entity)
- if get_current_task_context() is not None:
- await publish_task_event(
- "custom.checkpoint",
- message="Checkpoint reached",
- scope_key=f"tenant:{tenant_id}",
- actor=actor,
- entity=entity,
- )
- ```
-
- `QueueEventsConfig` groups live `delivery`, application `stream`, and durable `history`. It must enable at least one capability. Task events are separate from worker wakeups.
-
- - **Context-safe emission (`get_current_task_context()`):** Wrap `await context.progress(current=..., total=..., ...)` and `await context.log(...)` in helpers that check `if context is not None:` so `@task` functions execute identically inside queue workers and when called directly in unit tests or CLI commands. Pass `stage` (and optional `duration_ms`) inside `payload` so durable event history populates `QueueEventLogRecord.stage` and `duration_ms`.
- - **Attribution & entities:** Set attempt-scoped default attribution via `@task(..., actor=...)` (`QueueEventActor` or callable) or `TaskExecutionContext.actor`, and override `actor=`, `entity=` (`QueueEventEntityRef`), or `scope_key=` per `publish_task_event`, `publish_task_log`, or `publish_task_progress` call.
- - **Durable history queries & stage summaries:** Retrieve the active event log with `event_log = queue_service.get_event_log()`, query persisted events through `await event_log.query_events(QueueEventQuery(task_id=str(record.id), event_type="task.log", order="desc", limit=50), extra={"tenant_id": tenant_id})` (`OffsetPagination[QueueEventLogRecord]`), and aggregate stage durations with `await event_log.summarize_stages(QueueEventQuery(task_id=str(record.id)))` (`list[QueueEventStageSummary]`). Custom `QueueEventLog` implementations must define `publish_event_after_commit()` and `aclose()`.
- - **Browser SSE / WebSocket streams:** Consume `EventStreamConfig` endpoints from the browser with `createQueueEventStream` from `litestar-vite-plugin/helpers` (set `replay_limit > 0` on `EventStreamConfig` when reconnect replay is required; see [litestar-vite streams](../litestar-vite/references/streams.md)).
- - **Standalone event producer:** Use `async with create_event_producer(queue_config) as producer:` (alongside `bind_task_context` and `bind_beat_sink`) to emit task events from external processes without starting a full worker or Litestar app.
-
- ### Observability and Telemetry
-
- ```python
- 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
-
- ```python
- from litestar_queues import QueueConfig, QueueMaintenanceConfig
- from litestar_queues.events import QueueEventQuery, QueueEventRetentionRule
-
- queue_config = QueueConfig(
- queue_backend="sqlspec",
- maintenance=QueueMaintenanceConfig(
- time_budget=300,
- coordination_timeout=360,
- external_limit=100,
- stale_after=900,
- stale_limit=100,
- terminal_retention=30 * 24 * 60 * 60,
- terminal_limit=1000,
- event_retention_rules=(
- QueueEventRetentionRule(
- max_age=7 * 24 * 60 * 60,
- match=QueueEventQuery(event_type="task.log"),
- ),
- ),
- event_limit=1000,
- ),
- )
- ```
-
- ```bash
- 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 (`external_limit`, default `100`), stale recovery, terminal retention, then event retention (`event_retention_rules`). Custom CLI commands can also run phases programmatically via `await QueueMaintenanceService(service, maintenance_config).run(phases=["external", "stale"])` (`from litestar_queues.maintenance import QueueMaintenanceService`) and inspect `summary.outcome` (`"completed"`, `"already_running"`, `"partial"`, or `"failed"`). It never starts a worker, executes queued work, or loops to drain a backlog. Exhausting `time_budget` or reaching `external_limit` reports a `"partial"` outcome (`exit code 2`).
-
- Schedule one external six-hour or daily invocation. Retention phases have no destructive defaults: `stale_after`, `terminal_retention`, and `event_retention_rules` remain disabled when `None` or empty. `coordination_timeout` must exceed `time_budget`.
-
- ### External One-Task Execution
-
- ```bash
- QUEUES_TASK_ID=4d821c46-8c60-4ec3-b884-3f62eb71a03e \
- QUEUES_CONFIG_FACTORY=app.queue:create_queue_config \
- LITESTAR_QUEUES_TASK_MODULES=app.tasks \
- litestar queues run-task
- ```
-
- `run-task` claims and executes one existing record for an external executor (such as Cloud Run Jobs), reading `QUEUES_TASK_ID`, `QUEUES_CONFIG_FACTORY`, and `LITESTAR_QUEUES_TASK_MODULES` (or `<NAMESPACE>_TASK_MODULES`) by default. `--task-id`, `--config-factory`, and `--task-modules` override those environment variables for manual operation. It is not a standalone worker loop.
-
- ### Background Responses
-
- ```python
- from litestar import Response, post
- from litestar_queues import QueuedBackgroundTask, task
-
-
- @task("imports.process")
- async def process_import(path: str) -> None:
- pass
+ ### Events, Progress, Observability, and Bounded Maintenance
+ - **`QueueEventsConfig`**: Groups live `delivery` (`EventDeliveryConfig`), browser `stream` (`EventStreamConfig(path="/queues/events", transports={"sse", "websocket"})`), and durable `history` (`EventHistoryConfig` with optional `EventHistoryExtraColumn`). Inside `@task` functions, check `context = get_current_task_context()` before calling `await context.progress(...)`, `await context.log(...)`, or `await publish_task_event(...)`, and call `beat("stage")` for heartbeats. Query persisted events via `await queue_service.get_event_log().query_events(QueueEventQuery(...))` and `summarize_stages(...)`.
+ - **`ObservabilityConfig`**: Configures OpenTelemetry tracing (`enable_otel=True`), Prometheus metrics (`enable_prometheus=True`), and SQLCommenter (`enable_sqlcommenter=True`), plus `scheduler_canary_task` verified by `litestar queues scheduler-health --minutes 5`.
+ - **Bounded Maintenance (`QueueMaintenanceConfig`)**: Runs four bounded phases in fixed order (`external` reconciliation, `stale` recovery, `terminal` retention, `event` retention) via `litestar queues run-maintenance --json` or `QueueMaintenanceService(service, config).run()`. Retention thresholds (`stale_after`, `terminal_retention`, `event_retention_rules`) default to `None`/empty (disabled); `coordination_timeout` must exceed `time_budget`. Schedule one external invocation every six hours or daily. See [Backends, Worker Configuration, Events, and Maintenance](references/backends-and-events.md) for full configuration snippets.
- @post("/imports")
- async def create_import() -> Response[dict[str, str]]:
- return Response(
- {"status": "accepted"},
- background=QueuedBackgroundTask(process_import, "/tmp/data.csv"),
- )
- ```
+ ### External One-Task Execution & Background Responses
- `QueuedBackgroundTask` enqueues after the response is sent. It resolves the active plugin service at construction; pass `service=queue_service` for a custom service.
+ - **`litestar queues run-task`**: Claims and executes one existing record for an external executor (such as Cloud Run Jobs), reading `QUEUES_TASK_ID`, `QUEUES_CONFIG_FACTORY`, and `LITESTAR_QUEUES_TASK_MODULES` (or `--task-id`, `--config-factory`, `--task-modules`). It is not a standalone worker loop.
+ - **`QueuedBackgroundTask`**: Return `Response({"status": "accepted"}, background=QueuedBackgroundTask(process_import, "/tmp/data.csv"))` from a route handler to enqueue a task after the HTTP response is sent.
<workflow>
## 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, Cloud Tasks for serverless push delivery, 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.
+ 2. **Match the queue backend.** Use `"memory"` for same-process tests, `SQLSpecBackendConfig` for SQLSpec apps, `SQLAlchemyBackendConfig` for Advanced Alchemy apps, `RedisBackendConfig` for Redis, or `ValkeyBackendConfig` for Valkey.
+ 3. **Pick execution placement.** Use `"local"` workers by default, `"immediate"` for tests/scripts, broker consumers (Kafka, Pub/Sub, RabbitMQ, SQS) for distributed dispatch, `CloudTasksExecutionConfig` for serverless push delivery, and `CloudRunExecutionConfig` for isolated external jobs.
+ 4. **Configure the worker & storage.** Put worker settings under `WorkerConfig` and provision SQLSpec or Advanced Alchemy schema migrations explicitly.
+ 5. **Define, discover, and enqueue tasks.** Decorate callables with `@task`, import task modules via `task_modules` or `discover_tasks()` before string enqueueing or schedule init, and inject `NamedDependency[QueueService]` in route handlers.
+ 6. **Add optional capabilities & operational commands.** Configure worker wakeups, `QueueEventsConfig`, permanent uniqueness, `ObservabilityConfig`, and `QueueMaintenanceConfig` as needed; run `litestar queues run` / `run-consumer` continuously, `run-task` in one-shot executors, and `run-maintenance` on a six-hour or daily schedule.
</workflow>
<guardrails>
## Guardrails
- **Use `SQLAlchemyBackendConfig` for Advanced Alchemy persistence.** Import it from `litestar_queues.backends.advanced_alchemy` (do not look for `AdvancedAlchemyAsyncBackendConfig` or `AdvancedAlchemySyncBackendConfig`).
- - **Use current 0.12.0 config and status types.** Use `CloudRunExecutionConfig` (not `CloudRunExecutorConfig`), `TaskStatus` (not `JobStatus`), `RetryBackoff` (not `BackoffStrategy`), `ScheduleConfig` (not `Schedule`), and integer `priority` (not `TaskPriority`); do not look for removed `ConcurrencyLock` or `RateLimit` classes.
+ - **Use current config and status types.** Use `CloudRunExecutionConfig` (not `CloudRunExecutorConfig`), `TaskStatus` (not `JobStatus`), `RetryBackoff` (not `BackoffStrategy`), `ScheduleConfig` (not `Schedule`), and integer `priority` (not `TaskPriority`); do not look for removed `ConcurrencyLock` or `RateLimit` classes.
- **Use `queue_table_name` on `SQLSpecBackendConfig`.** Do not pass the removed `table_name` parameter.
- **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 or look for `MemoryBackendConfig`.** Pass `queue_backend="memory"` or `InMemoryQueueBackend`; 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()`.
- - **Do not name `@task` parameters after reserved `QueueService.enqueue` options.** `scheduled_at`, `run_after`, `expires_in`, `expires_at`, `key`, `queue`, `priority`, `retries`, `timeout`, `execution_backend`, `execution_profile`, `description`, `log_level`, `log_success`, `requeue_on_stale`, and `metadata` are consumed by `enqueue()` instead of reaching `record.kwargs`.
- - **Do not set both `task_dependency_resolver` and `task_dependency_provider`.** They are mutually exclusive on `QueueConfig`.
- - **Do not maintain separate `--enqueue` vs direct-call CLI code paths.** Route foreground CLI invocations through `await queue_service.enqueue(task, ..., execution_backend="immediate")` so persistence, execution, and terminal events share one pipeline.
</guardrails>
<validation>
## Validation Checkpoint
- - [ ] The dependency floor is `litestar-queues>=0.12.0`
+ - [ ] The dependency floor is `litestar-queues>=0.14.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]`
- - [ ] `@task` parameter names avoid the 16 reserved `QueueService.enqueue` keyword arguments
- - [ ] Only one of `task_dependency_resolver` or `task_dependency_provider` is configured
- [ ] 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
</validation>
<example>
## Example
**Task:** Persist report jobs with SQLSpec, run a standalone worker, deduplicate equivalent calls, and perform bounded retention.
```python
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)],
)
```
```bash
LITESTAR_APP=app:app litestar queues run
LITESTAR_APP=app:app litestar queues run-maintenance --json
```
</example>
## References Index
+ - **[Backends, Worker Configuration, Events, and Maintenance](references/backends-and-events.md)** — `task_dependency_provider`, Dishka worker scope, `create_queue_config`, SQLSpec/Advanced Alchemy/Redis/Valkey backend setups, `QueueEventsConfig`, `ObservabilityConfig`, and `QueueMaintenanceConfig`.
- **[Execution Backends](references/execution-backends.md)** — local, immediate, Cloud Run, Cloud Tasks, Kafka, Pub/Sub, RabbitMQ, SQS, delivery fencing, and lost-delivery repair.
- **[Namespacing and Expiry](references/namespacing.md)** — `namespace=`, not-started deadlines and the `expired` state, event-stream configuration.
## Cross-References
- **[litestar](../litestar/SKILL.md)** — Litestar app setup, plugin lists, DI, and lifespan.
- - **[Litestar routing](../litestar/references/handlers.md)** — Route handlers and controllers for enqueue endpoints.
- - **[Litestar DI](../litestar/references/di-and-dishka.md)** — `NamedDependency` and service injection.
- - **[Litestar plugins](../litestar/references/plugins.md)** — Plugin initialization and lifecycle.
- **[litestar-autowire](../litestar-autowire/SKILL.md)** — Optional task discovery through Autowire integration.
- - **[Litestar Channels & SSE](../litestar/references/channels-and-sse.md)** — Channels, SSE, WebSockets, and task-event fan-out.
- **[litestar-testing](../litestar-testing/SKILL.md)** — Application and handler tests.
- **[sqlspec](../sqlspec/SKILL.md)** — SQLSpec adapter and migration configuration.
- **[advanced-alchemy](../advanced-alchemy/SKILL.md)** — SQLAlchemy models, services, and Alembic ownership.
## Official References
- - <https://github.com/cofin/litestar-queues/tree/v0.12.0>
- - <https://github.com/cofin/litestar-queues/releases/tag/v0.12.0>
- - <https://github.com/cofin/litestar-queues/blob/v0.12.0/src/litestar_queues/config.py>
- - <https://github.com/cofin/litestar-queues/blob/v0.12.0/src/litestar_queues/task.py>
- - <https://github.com/cofin/litestar-queues/blob/v0.12.0/src/litestar_queues/_cli.py>
- - <https://github.com/cofin/litestar-queues/blob/v0.12.0/src/litestar_queues/backends/sqlspec/config.py>
- - <https://github.com/cofin/litestar-queues/blob/v0.12.0/src/litestar_queues/backends/advanced_alchemy/config.py>
- - <https://github.com/cofin/litestar-queues/blob/v0.12.0/src/litestar_queues/maintenance.py>
+ - <https://github.com/cofin/litestar-queues>
+ - <https://pypi.org/project/litestar-queues/>
## Shared Styleguide Baseline
- - Use shared styleguides for generic language/framework rules to reduce duplication in this skill.
- [General Principles](../litestar-styleguide/references/general.md)
- [Python](../litestar-styleguide/references/python.md)
- [Litestar](../litestar-styleguide/references/litestar.md)
- - Keep this skill focused on `litestar-queues` workflows, backend selection, worker placement, uniqueness, maintenance, and task-event APIs.