Skip to content

Python Getting Started

This guide takes you from uv init to a job reaching completed.

Version used in this guide

The install commands pin v0.6.6, the latest stable release. The canonical example is tested against both that release and the code on main; development-only 0.7 surfaces elsewhere on this site are identified by the site banner and stability labels.

Prerequisites

  • PostgreSQL running locally or remotely
  • uv (it will use or install a compatible Python 3.10+ interpreter)
  • A database URL exported as DATABASE_URL

Example local URL:

export DATABASE_URL=postgres://postgres:test@localhost:15432/awa_test

1. Create a Project

uv init awa-python-quickstart --bare
cd awa-python-quickstart
uv add awa-pg==0.6.6

uv init creates a minimal Python project. uv add creates and manages the project's virtual environment, records awa-pg in pyproject.toml, and writes a lockfile—there is no environment activation step.

2. Run Migrations

uv run python -m awa --database-url "$DATABASE_URL" migrate

3. Create a Worker

Create quickstart.py:

"""Awa Python quickstart — a complete runnable example.

Requires: uv add awa-pg==0.6.6
Requires: a running Postgres instance with DATABASE_URL set.

Usage from the repository's awa-python directory:
    DATABASE_URL=postgres://localhost/mydb uv run python examples/quickstart.py
"""

import asyncio
import os
from dataclasses import dataclass

import awa

DATABASE_URL = os.environ.get(
    "DATABASE_URL", "postgres://postgres:test@localhost:15432/awa_test"
)


@dataclass
class SendEmail:
    to: str
    subject: str


async def main():
    client = awa.AsyncClient(DATABASE_URL)
    await client.migrate()

    # Define a worker
    @client.task(SendEmail, queue="email")
    async def handle_email(job):
        print(f"Sending email to {job.args.to}: {job.args.subject}")

    # Start processing before the first enqueue so a fresh 0.6 database can
    # auto-finalize to the queue-storage engine.
    await client.start([("email", 2)])

    # Insert a job
    job = await client.insert(
        SendEmail(to="alice@example.com", subject="Welcome"),
        queue="email",
    )
    print(f"Inserted job {job.id} (kind={job.kind}, state={job.state})")

    # Verify it reaches a terminal state without relying on a fixed delay.
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 10
    last_state = job.state
    try:
        while True:
            remaining = deadline - loop.time()
            if remaining <= 0:
                raise TimeoutError(
                    f"timed out waiting for job {job.id} "
                    f"(last state: {last_state})"
                )

            # get_job is a single read-only query, so cancelling this await
            # cannot leave an application transaction partially committed.
            try:
                result = await asyncio.wait_for(
                    client.get_job(job.id), timeout=remaining
                )
            except asyncio.TimeoutError as error:
                raise TimeoutError(
                    f"timed out waiting for job {job.id} "
                    f"(last state: {last_state})"
                ) from error

            last_state = result.state
            if result.state == awa.JobState.Completed:
                break
            if result.state in (awa.JobState.Failed, awa.JobState.Cancelled):
                raise RuntimeError(
                    f"job {result.id} ended in terminal state {result.state}"
                )
            await asyncio.sleep(min(0.1, max(0, deadline - loop.time())))
    finally:
        await client.shutdown()
        await client.close()

    print(f"Job {result.id} state: {result.state}")


if __name__ == "__main__":
    asyncio.run(main())

This page includes the repository's canonical example verbatim. CI runs it against PostgreSQL and the docs check compiles its Python syntax.

4. Run It

uv run python quickstart.py

You should see the inserted job, the handler output, and the terminal state. The first two lines can swap order because the worker starts before the insert:

Inserted job 1 (kind=send_email, state=available)
Sending email to alice@example.com: Welcome
Job 1 state: completed

What happened?

  1. client.migrate() made the example standalone; the explicit migration command in step 2 is the deployment-friendly path.
  2. Inserting the job wrote durable state to PostgreSQL.
  3. The worker claimed it, incremented the attempt, and kept the claim alive while the handler ran.
  4. The handler result became durable completed state, which the final query read back.

Retries, callback waits, and progress checkpoints follow the same rule: PostgreSQL is the system of record, not worker memory. When you debug a job, inspect its durable snapshot first instead of relying only on worker logs.

5. Inspect the Queue

uv run python -m awa --database-url "$DATABASE_URL" job list --queue email
uv run python -m awa --database-url "$DATABASE_URL" job dump 1
uv run python -m awa --database-url "$DATABASE_URL" job dump-run 1
uv run python -m awa --database-url "$DATABASE_URL" queue stats

job dump gives you the whole job snapshot as JSON. job dump-run focuses on one attempt: the current attempt uses live row data, while historical attempts are reconstructed from the stored errors[] history.

6. Web UI (optional)

The dashboard ships in a separate wheel so the default awa-pg install stays small for workers and producers. Install the [ui] extra to bring in the awa-cli binary that hosts it:

uv add 'awa-pg[ui]==0.6.6'
uv run python -m awa --database-url "$DATABASE_URL" serve
# → http://127.0.0.1:3000

uv run python -m awa serve delegates to the awa serve binary (you can also call awa serve directly once the extra is installed). The UI is read-only when the database reports transaction_read_only = on (e.g. on a replica) or when --read-only is passed.

Useful Variants

  • await client.migrate() runs migrations from Python instead of the CLI.
  • awa.Client provides a synchronous API for worker/admin/direct-producer code — all methods are plain (e.g., client.insert(...), client.migrate()).
  • client.start() accepts tuple queue configs for hard-reserved mode and dict configs for weighted mode. See Configuration reference.
  • awa.PartitionedQueue helps one hot logical queue use several physical queues while keeping routing deterministic. See Partitioned queues.

ORM Transaction Bridging

Most applications should keep using their normal database stack for business tables. Use AsyncClient/Client for workers, admin calls, migrations, and queue-only producers; when a web request already has a transaction, enqueue through awa.bridge on that same connection/session.

Install the app database libraries you already use, for example:

uv add 'sqlalchemy[asyncio]' asyncpg

Then enqueue in the same SQLAlchemy transaction as your application write:

from dataclasses import dataclass

from awa.bridge import insert_job
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession


@dataclass
class SendEmail:
    to: str
    subject: str


async def create_order(session: AsyncSession, order_id: str, email: str) -> int:
    async with session.begin():
        await session.execute(
            text("INSERT INTO orders (id, email) VALUES (:id, :email)"),
            {"id": order_id, "email": email},
        )
        job = await insert_job(
            session,
            SendEmail(to=email, subject="Order confirmed"),
            queue="email",
        )
    return job["id"]

The same bridge supports asyncpg, psycopg3, SQLAlchemy, and Django; see Bridge Adapters for driver-specific examples.

By default, a queue is strict FIFO per (queue, priority). Operators can opt a contended queue into partitioned FIFO by raising awa.queue_meta.enqueue_shards — order is then preserved within each shard, but not across shards. If your producer enqueues jobs that must be processed in order (per-customer events, sequential workflow steps), pass ordering_key so they all land on one shard:

await client.insert(
    UpdateCustomer(customer_id=42, payload=...),
    queue="customer-updates",
    ordering_key=b"customer-42",  # str also accepted; UTF-8 encoded
)

At the default enqueue_shards = 1 the key is ignored (everything is on shard 0 anyway). See ADR-025 for the partitioned-FIFO contract and queue configuration for the operator-side knob.

Exporting OpenTelemetry metrics

awa records 20+ metrics (throughput, pickup latency, in-flight jobs, rescues, …) on the Rust side. Python workers enable OTLP export by calling awa.init_telemetry(...) once before the worker starts:

import os
import awa

awa.init_telemetry(
    os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"],   # e.g. http://localhost:4317
    os.environ.get("OTEL_SERVICE_NAME", "my-service"),
)
# ... then build the client and start workers as normal.

init_telemetry is idempotent; only the first call installs a provider. Call awa.shutdown_telemetry() at the end of short-lived scripts to flush pending metrics. See awa-python/examples/telemetry.py for a runnable example.

Distributed tracing

Distributed tracing (ADR-039) is automatic when opentelemetry-api is importable (install it directly, via your tracing stack, or with the awa-pg[otel] extra) — no configuration:

  • Producers: insert / insert_many_copy / enqueue_many_copy and the awa.bridge helpers capture the current OpenTelemetry span's context into the reserved awa:traceparent metadata key. An explicit metadata={"awa:traceparent": ...} always wins, and AWA_TRACE_CAPTURE=off disables ambient capture.
  • Handlers: the worker attaches the job's trace context as the ambient OpenTelemetry context before invoking your handler, so instrumented libraries (httpx, requests, SQLAlchemy, ...) and spans you create nest into the job's trace with no extra code. job.traceparent still exposes the stored enqueue-site value for inspection.

Without opentelemetry installed, all of this is a cached no-op.

To export awa's own Rust-side spans (send {queue}, job.execute {kind}) to your collector, init_telemetry now installs a trace pipeline alongside metrics (pass traces=False to keep it metrics-only). With it enabled, handler-side spans nest under the execution span; without it, they attach to the enqueue-site context so the trace still connects.

More Examples