litestar-org/litestar-skills/litestar-saq

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

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,
                ),
            ],
        ),
    )

from litestar import Litestar
from app.server.plugins import saq_plugin
app = Litestar(
    route_handlers=[...],
    plugins=[saq_plugin],
)

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)

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"}

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"}

if job is not None:
    await reports_service.repository.session.commit()
else:
    await reports_service.repository.session.rollback()

# 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

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