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]
• 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 (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()