Litestar Queues

Litestar Queues lets a Litestar application persist work, run it in a worker, and inspect the result. Use it for work that should outlive the request that started it: sending email, importing files, refreshing reports, or calling a slow service.

Quickstart

Install the package:

pip install litestar-queues

Create app.py:

from litestar import Litestar, post
from litestar.di import NamedDependency

from litestar_queues import QueueConfig, QueuePlugin, QueueService, task


@task("accounts.sync", queue="accounts", timeout=30)
async def sync_account(account_id: str) -> dict[str, str]:
    return {"account_id": account_id, "status": "synced"}


@post("/accounts/{account_id:str}/sync")
async def create_sync_job(
    account_id: str, queue_service: NamedDependency[QueueService]
) -> dict[str, str]:
    result = await queue_service.enqueue(sync_account, account_id)
    return {"task_id": str(result.id), "status": result.status or "pending"}


app = Litestar(
    route_handlers=[create_sync_job],
    plugins=[QueuePlugin(config=QueueConfig())],
)

Run the application:

LITESTAR_APP=app:app litestar run --reload

Enqueue a task:

curl -X POST http://127.0.0.1:8000/accounts/acct-123/sync

The response contains a task ID and an initial status:

{"task_id": "...", "status": "pending"}

By default, QueueConfig() starts a dedicated worker process alongside the server and uses an ephemeral SQLite database. The database is removed on normal shutdown and does not persist across restarts.

Production boundary

The default setup uses a temporary SQLite database and a local worker to get started quickly.

In production, decouple task storage from execution:

• Storage: Use Redis, Valkey, SQLSpec, or Advanced Alchemy so tasks persist across deploys and restarts.
• Workers: Run standalone worker processes or Cloud Run jobs so background work scales independently from web requests.

Next steps

• Start here
• Understand the model
• Follow a how-to guide
• Choose backends
• Run an example
• Browse the API reference

Litestar Queues supports Python 3.10 through 3.14 and is licensed under MIT.