Litestar Saq

litestar-saq

litestar-saq is the first-party plugin that integrates SAQ (Simple Async Queue) with Litestar. It provides:

• SAQPlugin — registers queues, workers, lifespan management, and DI for TaskQueues
• SAQConfig / QueueConfig — declarative plugin, queue, worker, broker, shutdown, polling, and OpenTelemetry configuration
• litestar workers run — CLI to start worker processes, optionally filtered by queue
• Optional web UI mounted under the Litestar app
• DI injection of TaskQueues into route handlers for ergonomic enqueueing

Code Style Rules

• Use PEP 604 unions: T | None, never Optional[T]
• Consumer Litestar app modules MAY use from __future__ import annotations — canonical Litestar apps do.
• Async all I/O — task bodies and enqueue calls are async def.
• First positional arg of every task is ctx: dict (the SAQ context dict).
• Task params after * are keyword-only.
• Use NamedDependency[TaskQueues] for handler injection. TaskQueues is registered under the task_queues dependency key, and Litestar 2.24 deprecates implicit DI.

Quick Reference

Plugin Setup (canonical pattern)

The canonical pattern from litestar-fullstack (src/py/app/server/plugins.py) uses lazy initialization and use_server_lifespan=True so worker child processes start and stop with the Litestar server lifespan:

# Branch A — SAQ with Redis as the broker (pick when Redis is already in-stack
# for cache / sessions, or when you want the SAQ web UI + multi-queue fanout).
from litestar_saq import SAQConfig, SAQPlugin, QueueConfig, CronJob

from app.lib.settings import get_settings

def create_saq_plugin() -> SAQPlugin:
    settings = get_settings()
    return SAQPlugin(
        config=SAQConfig(
            use_server_lifespan=True,            # worker child processes follow server lifespan
            web_enabled=settings.saq.web_enabled,
            enable_otel=None,                    # auto-detect if OpenTelemetry is installed and configured
            queue_configs=[
                QueueConfig(
                    name="default",
                    dsn=settings.redis.url,      # redis://... — Redis broker
                    tasks=["app.domain.system.tasks.send_email"],
                    scheduled_tasks=[
                        CronJob(
                            function="app.domain.system.tasks.cleanup_sessions",
                            cron="*/15 * * * *",
                            timeout=120,
                        ),
                    ],
                ),
            ],
        ),
    )

saq_plugin = create_saq_plugin()

# Branch B — SAQ with PostgreSQL as the broker (install `litestar-saq[psycopg]`;
# pick when the project is
# PG-only, you want one less piece of infra, or throughput is moderate).
def create_saq_plugin_pg() -> SAQPlugin:
    settings = get_settings()
    return SAQPlugin(
        config=SAQConfig(
            use_server_lifespan=True,
            web_enabled=settings.saq.web_enabled,
            queue_configs=[
                QueueConfig(
                    name="default",
                    dsn=settings.database.url,   # postgresql://... — PG broker
                    tasks=["app.domain.system.tasks.send_email"],
                ),
            ],
        ),
    )

Pick Branch A (SAQ + Redis) when: Redis is already in-stack (cache, sessions, Channels), you need multi-queue fanout across many workers, want the SAQ web UI and dead-letter dashboards, or have high throughput (>1k jobs/s per queue).

Pick Branch B (SAQ + PostgreSQL) when: PG-only deployment (Cloud SQL, AlloyDB, self-hosted single DB), avoiding Redis, durable SQL-backed SAQ storage, SQL-queryable job history, or moderate throughput ( None: """Purge soft-deleted records every night at 02:00 UTC.""" ...

@task(priority=5, retries=1, timeout=300, executiontarget="cloudrun") async def generatereport(*, report_id: int) -> None: """Export report — runs on Cloud Run for isolation.""" ...

Enqueue imperatively (from a handler or service):

await generatereport.enqueue(executiontarget="cloudrun", report_id=42)

### Wire into Litestar

```python
from litestar import Litestar
from app.server.plugins import saq_plugin

app = Litestar(
    route_handlers=[...],
    plugins=[saq_plugin],
)

Define a Task

# app/domain/system/tasks.py
async def send_email(ctx: dict, *, recipient: str, subject: str, body: str) -> None:
    """Send an email as a background job.

    Args:
        ctx: SAQ context dict populated by worker hooks.
        recipient: To address.
        subject: Email subject.
        body: Email body.
    """
    email_service = ctx["email_service"]
    await email_service.send(recipient, subject, body)

For long-running work, decorate the task with monitored_job() so heartbeats are sent while the task runs:

from litestar_saq import monitored_job

@monitored_job()
async def rebuild_index(ctx: dict, *, index_name: str) -> dict[str, str]:
    await run_rebuild(index_name)
    return {"status": "complete"}

Enqueue from a Handler (DI of TaskQueues)

from litestar import Controller, post
from litestar.di import NamedDependency
from litestar_saq import TaskQueues

class NotificationController(Controller):
    path = "/api/notifications"

    @post("/")
    async def queue_notification(
        self,
        data: NotificationCreate,
        task_queues: NamedDependency[TaskQueues],
    ) -> dict[str, str]:
        queue = task_queues.get("default")
        await queue.enqueue(
            "send_email",
            recipient=data.email,
            subject=data.subject,
            body=data.body,
            timeout=30,
            retries=2,
            key=f"notify-{data.email}",
        )
        return {"status": "queued"}

CLI

# Run workers (uses the same Litestar app)
litestar --app app:app workers run

# Run multiple worker processes
litestar --app app:app workers run --workers 4

# Run only selected queues in this worker service
litestar --app app:app workers run --queues emails --queues reports

# Inspect queues
litestar --app app:app workers status

Web UI

When web_enabled=True, the SAQ web UI is mounted under the Litestar app for queue introspection and job retry.

Job Options

| Option | Default | Use | | --- | --- | --- | | timeout | 10 | Always set explicitly — SAQ's default is usually too low or too high for real jobs | | retries | 1 | Retry count on exception | | ttl | 600 | Seconds to retain result after completion | | key | None | Deduplication key — skip if already queued | | heartbeat | 0 | Heartbeat interval for long-running jobs | | scheduled | 0 | Unix timestamp to delay start |

Workflow

Step 1: Install

pip install litestar-saq
pip install "litestar-saq[psycopg]"  # PostgreSQL broker
pip install "litestar-saq[otel]"     # OpenTelemetry spans

Step 2: Define Queues

Build QueueConfig instances for each logical queue ("default", "emails", "reports"). Put the broker dsn or broker_instance on each QueueConfig. Reference task functions by dotted path or callable; the plugin imports dotted paths at startup.

Step 3: Configure Plugin

Wrap QueueConfigs in SAQConfig. Pick Redis (redis://...) when Redis is already in the stack; pick PostgreSQL (postgresql://...) when the project is PG-only. HTTP queues are supported for delegating to a remote SAQ service. Set use_server_lifespan=True when the web process should own worker child processes. Toggle web_enabled / web_guards for the introspection UI and enable_otel for tracing.

Step 4: Define Tasks

Place task functions in app/domain//tasks.py. First arg ctx: dict, rest keyword-only. Add shared resources (DB, HTTP client, email service) in QueueConfig.startup / before_process hooks and read them from ctx.

Step 5: Schedule Cron Work

Add CronJob entries to QueueConfig.scheduled_tasks for recurring work. Always set timeout. Do not use external cron tools for work that belongs in the queue.

Step 6: Enqueue from Handlers

Inject TaskQueues into route handlers. Use task_queues.get("name") then await queue.enqueue("task_name", ...). Use key= for deduplication.

Step 7: Publish to Channels (optional)

For real-time updates after a job completes, publish to Litestar Channels from inside the task. See ../litestar-realtime/references/websockets.md.

Step 8: Run

For dev: litestar run can start worker child processes when use_server_lifespan=True. For production: run litestar workers run --workers N as a separate service/process from litestar run. There is no --process flag in litestar-saq 0.8.0.

Guardrails

• Use litestar-saq, not raw SAQ, in Litestar apps — the plugin handles DI, lifespan, CLI, and the web UI. Raw SAQ misses all of that.
• Always set timeout on tasks and CronJobs — SAQ defaults to 10s, which is rarely the correct production value.
• Use monitored_job() or heartbeat for jobs that run longer than ~30s, otherwise SAQ may mark them stuck and re-queue.
• Inject TaskQueues via DI — don't import a global queue inside handlers. The plugin owns the queue lifecycle.
• Use CronJob for scheduled work — not external cron. CronJobs participate in retries, timeouts, and observability.
• Use key= for deduplication — same logical job (per-user sync, per-resource refresh) should not stack.
• use_server_lifespan=True for dev and small-to-mid apps that should start worker child processes with the web server. For high-throughput production, run litestar workers run --workers N as a separate service.
• Use dsn for multi-process workers — Python 3.14 forkserver/spawn support rebuilds brokers in child processes from QueueConfig.dsn; broker_instance-only queues cannot be pickled into child workers.
• Set graceful shutdown controls for long jobs — use shutdown_grace_period_s and, when needed, cancellation_hard_deadline_s on QueueConfig.
• Publish to Litestar Channels from tasks when the job result must update connected websocket clients. See ../litestar-realtime/references/websockets.md.
• Pull shared resources from ctx populated by QueueConfig hooks, not module-level globals — keeps tests deterministic and supports per-worker init.
• Reach for the sidecar worker pattern when you need same-transaction outbox semantics, a project-owned job schema, batched heartbeats for many running jobs, frontend updates through channels, or execution-target routing across Cloud Run / local. Keep the stack explicit: TaskService for fenced SQL transitions, Worker for execution, WorkerSidecar for wakeups/batched heartbeats/channel publishing, and WorkerPlugin for Litestar lifecycle wiring. For normal PG-backed queueing, SAQ+PG is the simpler default. See [references/postgres-native-sidecar-worker.md](references/postgres-native-sidecar-worker.md).

Validation Checkpoint

Before delivering Litestar + SAQ code, verify:

• [ ] SAQPlugin is in app.plugins
• [ ] SAQConfig.use_server_lifespan is set explicitly
• [ ] SAQConfig.worker_processes or CLI --workers is set intentionally
• [ ] Each QueueConfig has a broker dsn or broker_instance
• [ ] Multi-process worker configs use dsn, not broker_instance only
• [ ] Each QueueConfig lists tasks by dotted path; the imports resolve
• [ ] All tasks have ctx: dict as the first positional arg, keyword-only params after *
• [ ] Every task has timeout set
• [ ] Long-running jobs (>30s) have heartbeat set
• [ ] Long-running task functions use monitored_job() when they need automatic heartbeats
• [ ] CronJobs have timeout and a sensible cron expression
• [ ] Handlers enqueue via injected TaskQueues, not module globals
• [ ] Job dedup uses key= where applicable
• [ ] Production deploys run workers as a separate service (litestar workers run --workers N)

Example

Task: A Litestar app with a default queue, an email task, a cleanup CronJob, and a handler that enqueues notifications. This example uses Redis as the SAQ broker; swap dsn=settings.redis.url for dsn=settings.database.url if the project is PG-only — see Quick Reference above for both patterns.

# app/server/plugins.py
from litestar_saq import SAQConfig, SAQPlugin, QueueConfig, CronJob

from app.lib.settings import get_settings

def create_saq_plugin() -> SAQPlugin:
    settings = get_settings()
    return SAQPlugin(
        config=SAQConfig(
            use_server_lifespan=True,
            web_enabled=settings.saq.web_enabled,
            queue_configs=[
                QueueConfig(
                    name="default",
                    dsn=settings.redis.url,  # Redis broker — swap for settings.database.url in PG-only stacks
                    startup="app.domain.system.tasks.worker_startup",
                    shutdown="app.domain.system.tasks.worker_shutdown",
                    tasks=[
                        "app.domain.system.tasks.send_email",
                        "app.domain.system.tasks.cleanup_sessions",
                    ],
                    scheduled_tasks=[
                        CronJob(
                            function="app.domain.system.tasks.cleanup_sessions",
                            cron="*/15 * * * *",
                            timeout=120,
                        ),
                    ],
                ),
            ],
        ),
    )

saq_plugin = create_saq_plugin()

# app/domain/system/tasks.py
async def worker_startup(ctx: dict) -> None:
    """Initialize shared resources for this worker."""
    ctx["email_service"] = create_email_service()
    ctx["db"] = create_database_client()

async def worker_shutdown(ctx: dict) -> None:
    """Dispose shared worker resources."""
    await ctx["db"].close()

async def send_email(ctx: dict, *, recipient: str, subject: str, body: str) -> None:
    """Send an email as a background job."""
    email = ctx["email_service"]
    await email.send(recipient, subject, body)

async def cleanup_sessions(ctx: dict) -> None:
    """Purge expired sessions every 15 minutes."""
    db = ctx["db"]
    await db.execute("DELETE FROM session WHERE expires_at  dict[str, str]:
        queue = task_queues.get("default")
        await queue.enqueue(
            "send_email",
            recipient=data.email,
            subject=data.subject,
            body=data.body,
            timeout=30,
            retries=2,
            key=f"notify-{data.email}",
        )
        return {"status": "queued"}

# app.py
from litestar import Litestar

from app.domain.notifications.controllers import NotificationController
from app.server.plugins import saq_plugin

app = Litestar(
    route_handlers=[NotificationController],
    plugins=[saq_plugin],
)

# Dev: workers start with the Litestar server lifespan
litestar --app app:app run

# Prod: separate worker service/process
litestar --app app:app workers run --workers 4

---

References Index

• [Advanced Patterns](references/patterns.md) — Heartbeat tuning, dead-letter handling, job chaining, queue priorities, worker lifecycle hooks, Postgres backend.
• [Sidecar Worker Pattern](references/postgres-native-sidecar-worker.md) — TaskService + Worker + WorkerSidecar + WorkerPlugin pattern for same-transaction outbox semantics, project-owned job schema, sidecar batched heartbeats/wakeups, channel publish-back to the frontend, @task decorator + ScheduleConfig cron registry, and execution_target routing (local / cloudrun / immediate).

Cross-References

• [litestar](../litestar/SKILL.md) — Litestar app initialization, plugins, and lifespan.
• **[litestar websockets refer

…

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: litestar-org
• Source: litestar-org/litestar-skills
• License: MIT
• Homepage: https://github.com/litestar-org/litestar-skills

Install and usage instructions live in the source repository linked above.