/litestar-saq

name: litestar-saq
description: "Use when working with litestar_saq, SAQPlugin, SAQConfig, QueueConfig, TaskQueues, CronJob, litestar workers run, background jobs, schedules, or SAQ UI. Not for litestar-queues, Celery, RQ, or Dramatiq."

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: Context` (from `saq.types`, not `litestar_saq.typing.Context` which is the OpenTelemetry context) or `ctx: dict[str, Any]`.
• 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=settings.saq.otel_enabled,
            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. When sharing a database URL with SQLAlchemy (`postgresql+psycopg://` or `postgresql+asyncpg://`), strip the driver suffix so `QueueConfig.dsn` starts with `postgresql://`:

from litestar_saq import (
    CronJob,
    PostgresQueueOptions,
    QueueConfig,
    SAQConfig,
    SAQPlugin,
    after_process_logger,
    before_process_logger,
    shutdown_logger,
    startup_logger,
    timing_after_process,
    timing_before_process,
)

from app.lib.settings import get_settings
from app.lib.worker import after_process, before_process, on_shutdown, on_startup


def create_saq_plugin_pg() -> SAQPlugin:
    settings = get_settings()
    dsn = settings.db.url.replace("postgresql+psycopg", "postgresql").replace("postgresql+asyncpg", "postgresql")
    return SAQPlugin(
        config=SAQConfig(
            use_server_lifespan=settings.saq.use_server_lifespan,
            worker_processes=settings.saq.processes,
            web_enabled=settings.saq.web_enabled,
            enable_otel=settings.saq.otel_enabled,
            queue_configs=[
                QueueConfig(
                    name="background-tasks",
                    dsn=dsn,
                    concurrency=settings.saq.concurrency,
                    broker_options=PostgresQueueOptions(
                        jobs_table="task_queue",
                        stats_table="task_queue_stats",
                        versions_table="task_queue_ddl_version",
                        manage_pool_lifecycle=True,
                    ),
                    tasks=[
                        "app.domain.reports.tasks.generate_report",
                        "app.domain.reports.tasks.reap_abandoned_reports",
                    ],
                    scheduled_tasks=[
                        CronJob(
                            function="app.domain.reports.tasks.reap_abandoned_reports",
                            cron="0 */3 * * *",
                            timeout=60,
                        ),
                    ],
                    startup=[startup_logger, on_startup],
                    shutdown=[shutdown_logger, on_shutdown],
                    before_process=[timing_before_process, before_process_logger, before_process],
                    after_process=[after_process, timing_after_process, after_process_logger],
                    shutdown_grace_period_s=60,
                    cancellation_hard_deadline_s=10,
                ),
            ],
        ),
    )

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://` (`rediss://`, `redis+unix://`), `postgresql://`, or `http://` `dsn`, or a supported `broker_instance`. Supplying both or neither raises `ImproperlyConfiguredException`. PostgreSQL DSNs must start with `postgresql://` (`postgres://` raises `ImproperlyConfiguredException`). PostgreSQL requires `litestar-saq[psycopg]`; configure queue behavior with `bro