Skip to content

Quickstart

This guide walks through creating a runnable Litestar application that defines a background task, enqueues work from an HTTP route handler, and inspects task results.

Define a Task and Register the Plugin

Create app.py with the following code:

app.py
"""Minimal runnable Litestar Queues quickstart application."""

from uuid import UUID

from litestar import Litestar, get, post
from litestar.di import NamedDependency
from litestar.exceptions import NotFoundException
from litestar.params import FromPath, FromQuery

from litestar_queues import QueueConfig, QueuePlugin, QueueService, task


@task(queue="reports")
async def render_report(report_id: str, requested_by: str) -> dict[str, str]:
    """Render an analytics report in the background."""
    return {"report_id": report_id, "requested_by": requested_by, "status": "ready"}


@post("/reports/{report_id:str}")
async def start_report(
    report_id: FromPath[str],
    queue_service: NamedDependency[QueueService],
    requested_by: FromQuery[str] = "team@example.com",
) -> dict[str, str]:
    """Enqueue a background report job and return its tracking ID."""
    handle = await queue_service.enqueue(render_report, report_id, requested_by=requested_by)
    return {"task_id": str(handle.id), "status": handle.status or "pending"}


@get("/tasks/{task_id:uuid}")
async def get_report_status(task_id: FromPath[UUID], queue_service: NamedDependency[QueueService]) -> dict[str, object]:
    """Inspect the status and return value of a queued task."""
    record = await queue_service.get_task(task_id)
    if record is None:
        raise NotFoundException(detail=f"Task {task_id} not found")
    return {
        "task_id": str(record.id),
        "status": record.status,
        "is_terminal": record.is_terminal,
        "result": record.result,
    }


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

What Happens Under the Hood

  1. @task(queue="reports") wraps render_report in a Task bound to the "reports" queue and registers it in the task registry while preserving its direct async callability.
  2. QueueConfig() defaults to queue_backend="ephemeral" (a zero-config, cross-process WAL SQLite database) and WorkerConfig(placement="server").
  3. QueuePlugin registers QueueService for dependency injection (queue_service: NamedDependency[QueueService]) and hooks into the Litestar CLI so litestar run automatically supervises a dedicated worker subprocess.

Run the Application

Start the development server using the Litestar CLI:

litestar --app app:app run --reload

When litestar run starts with WorkerConfig(placement="server"), ServerWorkerSupervisor spawns a child litestar queues run process that claims work across all queues.

Enqueue a Task and Inspect the Result

In another terminal, enqueue a report job via POST /reports/q4-revenue:

curl -X POST "http://127.0.0.1:8000/reports/q4-revenue?requested_by=analytics@example.com"

{
  "task_id": "2c8c9a14-6d3c-4f2b-9f1a-8e3e4d7a1b20",
  "status": "pending"
}

Query the task status and return value using the returned task_id:

curl "http://127.0.0.1:8000/tasks/2c8c9a14-6d3c-4f2b-9f1a-8e3e4d7a1b20"

{
  "task_id": "2c8c9a14-6d3c-4f2b-9f1a-8e3e4d7a1b20",
  "status": "completed",
  "is_terminal": true,
  "result": {
    "report_id": "q4-revenue",
    "requested_by": "analytics@example.com",
    "status": "ready"
  }
}

Next Steps