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. QueueConfig() starts one fresh queue-worker child for this litestar run invocation and shares a private temporary SQLite file with it. No queue socket or port is exposed, and the temporary queue is removed on normal shutdown; it is not durable across server restarts.

Production boundary

Choose where tasks are stored separately from where they run. For durable deployments use a shared backend such as SQLSpec, Advanced Alchemy, Redis, or Valkey. Use standalone workers when the web app and task workers must scale separately. The process-local memory backend remains useful for inline tests or an explicitly single-ASGI-process worker. Cloud Run runs tasks; it does not store them.

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.