/litestar-saq

name: litestar-saq
description: "Auto-activate for litestar_saq, SAQPlugin, SAQConfig, QueueConfig, TaskQueues, CronJob, litestar workers run, background jobs, schedules, or SAQ web UI. Not for Celery, RQ, or Dramatiq — use their respective integrations or litestar-queues."

litestar-saq

`litestar-saq` is the first-party plugin that integrates [SAQ (Simple Async Queue)](https://github.com/tobymao/saq) 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]`
• 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).
• Pass job payload as keyword arguments so task signatures and enqueue calls stay explicit.
• 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](https://github.com/litestar-org/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.

**Redis broker:** Use this when Redis is already in the stack or is the chosen SAQ backend:

from litestar_saq import CronJob, QueueConfig, SAQConfig, SAQPlugin

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,
            enable_otel=None,
            queue_configs=[
                QueueConfig(
                    name="default",
                    dsn=settings.redis.url,
                    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()

**PostgreSQL broker:** Install `litestar-saq[psycopg]` when PostgreSQL is the chosen backend:

from litestar_saq import QueueConfig, SAQConfig, SAQPlugin

from app.lib.settings import get_settings


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,
                    tasks=["app.domain.system.tasks.send_email"],
                ),
            ],
        ),
    )

Choose the broker already supported by the deployment. PostgreSQL job writes use SAQ's own pool and transaction; they are not automatically atomic with writes made through an application ORM or SQL session.

Each `QueueConfig` accepts exactly one connection source: a supported `redis://`, `postgresql://`, or `http://` `dsn`, or a supported `broker_instance`. Supplying both or neither raises `ImproperlyConfiguredException`. PostgreSQL requires `litestar-saq[psycopg]`; configure queue behavior with `broker_options` and connection/client construction with `broker_instance_options`.

Wire into Litestar

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

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

Define a Task

Task functions live in `app/domain/<domain>/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, set the job's `heartbeat` stale threshold and decorate the task with `monitored_job()` so the plugin signals its batched `HeartbeatManager` while the task runs. `heartbeat` is not an update interval: SAQ marks an active job stuck when its last touch is older than that threshold.

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")
        job = 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" if job is not None else "duplicate"}

CLI

# Run workers (uses the same Litesta