Quickstart#

This first application stores work in memory and runs it with the worker that starts inside the Litestar process. You can replace both choices later without changing the task function or route.

Install Litestar Queues#

pip install litestar-queues

Create app.py#

Copy this complete file:

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 and verify it#

Start Litestar:

LITESTAR_APP=app:app litestar run --reload

In another terminal, enqueue the task:

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

You receive JSON shaped like this:

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

The ID identifies the saved task record. pending is its status when it enters the queue. The server-owned worker child may complete the task immediately after the response is created. The default private SQLite queue belongs to this server invocation and is removed when it stops.

Next steps#

  • Concepts explains records, workers, and backend choices.

  • Results shows how to refresh and wait for a result.

  • Run workers explains server, ASGI, and standalone placement.

  • Choose backends selects durable shared storage for production.

  • Task events publishes progress for applications and operators.