awa-pg 0.4.1

Postgres-native background job queue — Python SDK with async/sync workers, transactional enqueue, progress tracking, and web UI

Awa

Postgres-native background job queue for Rust and Python.

Awa (Māori: river) provides durable, transactional job enqueueing with typed handlers in both Rust and Python. All queue state lives in Postgres — no Redis, no RabbitMQ. The Rust runtime handles polling, heartbeating, crash recovery, and dispatch. Python workers run on that same runtime via PyO3, getting Rust-grade reliability with Python-native ergonomics.

Features

• Postgres-only — one dependency you already have.
• Transactional enqueue — insert jobs inside your business transaction. Commit = visible. Rollback = gone. Works with Awa's own transactions, your existing psycopg3/asyncpg/SQLAlchemy/Django connection (Python), or tokio-postgres (Rust).
• Cancel by unique key — cancel scheduled jobs by their insert-time components (kind + args) without storing job IDs.
• Rust and Python workers — same queues, identical semantics, mixed deployments.
• Crash recovery — heartbeat + hard deadline rescue. Stale jobs recovered automatically.
• Web UI — dashboard, job inspector, queue management, cron controls.
• Structured progress — handlers report percent, message, and checkpoint metadata; persisted across retries.
• Periodic/cron jobs — leader-elected scheduler with timezone support and atomic enqueue.
• Webhook callbacks — park jobs for external completion with optional CEL expression filtering.
• LISTEN/NOTIFY wakeup — sub-10ms pickup latency.
• OpenTelemetry — 20 built-in metrics (counters, histograms, gauges) for Prometheus/Grafana.
• Hot/cold storage — runnable work in a hot table, deferred work in a cold table.
• Rate limiting — per-queue token bucket. Weighted concurrency — global worker pool with per-queue guarantees.

Recent engineering benchmarks range from ~5k Python hot-path jobs/sec on a laptop to ~40k immediately-available Rust inserts/sec on dedicated Postgres hardware, with 4-producer enqueue contention runs above ~100k chunked INSERT/sec. See benchmarking notes for methodology, caveats, and the latest COPY-vs-INSERT results.

Core concurrency invariants (no duplicate processing after rescue, stale completions rejected, shutdown drain ordering) are checked with TLA+ models covering single and multi-instance deployments.

Getting Started

# 1. Install
pip install awa-pg awa-cli     # Python
# or: cargo add awa             # Rust

# 2. Start Postgres and run migrations
awa --database-url $DATABASE_URL migrate

# 3. Write a worker and start processing (see examples below)

# 4. Monitor
awa --database-url $DATABASE_URL serve   # → http://127.0.0.1:3000

Language-specific guides:

• Rust getting started
• Python getting started

Python Example

import awa
import asyncio
from dataclasses import dataclass

@dataclass
class SendEmail:
    to: str
    subject: str

async def main():
    client = awa.AsyncClient("postgres://localhost/mydb")
    await client.migrate()

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

    await client.insert(
        SendEmail(to="[e-mail]", subject="Welcome"),
        queue="email",
    )

    client.start([("email", 2)])
    await asyncio.sleep(1)
    await client.shutdown()

asyncio.run(main())

Progress tracking — checkpoint and resume on retry:

@client.worker(BatchImport, queue="etl")
async def handle_import(job):
    last_id = (job.progress or {}).get("metadata", {}).get("last_id", 0)
    for item in fetch_items(after=last_id):
        process(item)
        job.set_progress(50, "halfway")
        job.update_metadata({"last_id": item.id})
    await job.flush_progress()

Transactional enqueue — atomic with your business logic:

async with await client.transaction() as tx:
    await tx.execute("INSERT INTO orders (id) VALUES ($1)", order_id)
    await tx.insert(SendEmail(to="[e-mail]", subject="Order confirmed"))

Sync API for Django/Flask — use awa.Client for sync frameworks; all methods are plain (no suffix):

client = awa.Client("postgres://localhost/mydb")
client.migrate()
job = client.insert(SendEmail(to="[e-mail]", subject="Hello"))

ORM transaction bridging — insert jobs within your existing psycopg3, asyncpg, SQLAlchemy, or Django transaction. No separate Awa connection needed:

from awa.bridge import insert_job_sync

# Django
with transaction.atomic():
    Order.objects.create(...)
    insert_job_sync(connection, SendEmail(to="[e-mail]", subject="Order confirmed"))

# SQLAlchemy
with Session(engine) as session, session.begin():
    session.execute(...)
    insert_job_sync(session, SendEmail(to="[e-mail]", subject="Order confirmed"))

See examples/python/ for complete runnable scripts tested in CI.

Rust Example

use awa::{Client, QueueConfig, JobArgs, JobResult, JobError, JobContext, Worker};
use serde::{Serialize, Deserialize};

#[derive(Debug, Serialize, Deserialize, JobArgs)]
struct SendEmail {
    to: String,
    subject: String,
}

struct SendEmailWorker;

#[async_trait::async_trait]
impl Worker for SendEmailWorker {
    fn kind(&self) -> &'static str { "send_email" }

    async fn perform(&self, ctx: &JobContext) -> Result<JobResult, JobError> {
        let args: SendEmail = serde_json::from_value(ctx.job.args.clone())
            .map_err(|e| JobError::terminal(e.to_string()))?;
        send_email(&args.to, &args.subject).await
            .map_err(JobError::retryable)?;
        Ok(JobResult::Completed)
    }
}

// Insert a job (with uniqueness)
awa::insert_with(&pool, &SendEmail { to: "[e-mail]".into(), subject: "Welcome".into() },
    awa::InsertOpts { unique: Some(awa::UniqueOpts { by_args: true, ..Default::default() }), ..Default::default() },
).await?;

// Cancel by unique key (e.g., when the triggering condition is resolved)
awa::admin::cancel_by_unique_key(&pool, "send_email", None, Some(&serde_json::json!({"to": "[e-mail]", "subject": "Welcome"})), None).await?;

// Start workers with a typed lifecycle hook
let client = Client::builder(pool)
    .queue("default", QueueConfig::default())
    .register_worker(SendEmailWorker)
    .on_event::<SendEmail, _, _>(|event| async move {
        if let awa::JobEvent::Exhausted { args, error, .. } = event {
            tracing::error!(to = %args.to, error = %error, "email job exhausted retries");
        }
    })
    .build()?;
client.start().await?;

tokio-postgres bridge — insert jobs within existing tokio-postgres transactions (see bridge adapters):

use awa::bridge::tokio_pg; // requires features = ["tokio-postgres"]

// pg_client is a &mut tokio_postgres::Client (not the Awa Client above)
let txn = pg_client.transaction().await?;
txn.execute("INSERT INTO orders ...", &[&order_id]).await?;
tokio_pg::insert_job(&txn, &SendEmail { to: "[e-mail]".into(), subject: "Confirmed".into() }).await?;
txn.commit().await?;

Installation

Python

pip install awa-pg       # SDK: insert, worker, admin, progress
pip install awa-cli      # CLI: migrations, queue admin, web UI

Rust

[dependencies]
awa = "0.4"

CLI

Available via pip (no Rust toolchain needed) or cargo:

pip install awa-cli
# or: cargo install awa-cli

awa --database-url $DATABASE_URL migrate
awa --database-url $DATABASE_URL serve
awa --database-url $DATABASE_URL queue stats
awa --database-url $DATABASE_URL job list --state failed

Architecture

 ┌────────────────┐  ┌────────────────┐
 │ Rust producer  │  │  Python (pip)  │
 └───────┬────────┘  └────────┬───────┘
         └────────┬───────────┘
                  ▼
       ┌────────────────────┐
       │     PostgreSQL     │
       │     jobs_hot       │
       │     scheduled_jobs │
       └─────────┬──────────┘
                 │
       ┌─────────┼─────────┐
       ▼         ▼         ▼
   ┌────────┐┌────────┐┌────────┐
   │ Worker ││ Worker ││ Worker │
   │ (Rust) ││ (PyO3) ││ (PyO3) │
   └────────┘└────────┘└────────┘

All coordination through Postgres. The Rust runtime owns polling, heartbeats, shutdown, and crash recovery for both languages. Mixed Rust and Python workers coexist on the same queues. See architecture overview for full details.

Workspace

Crate	Purpose
awa	Main crate — re-exports awa-model + awa-worker
awa-model	Types, queries, migrations, admin ops
awa-macros	#[derive(JobArgs)] proc macro
awa-worker	Runtime: dispatch, heartbeat, maintenance
awa-ui	Web UI (axum API + embedded React frontend)
awa-cli	CLI binary (migrations, admin, serve)
awa-python	PyO3 extension module (pip install awa-pg)
awa-testing	Test helpers (TestClient)

Documentation

Doc	Description
Rust getting started	From cargo add to a job reaching completed
Python getting started	From pip install to a job reaching completed
Deployment guide	Docker, Kubernetes, pool sizing, graceful shutdown
Migration guide	Fresh installs, upgrades, extracted SQL, rollback strategy
Bridge adapters	tokio-postgres, psycopg3, asyncpg, SQLAlchemy, Django bridging
Configuration reference	QueueConfig, ClientBuilder, Python start(), env vars
Troubleshooting	Stuck running jobs, leader delays, heartbeat timeouts
Architecture overview	System design, data flow, state machine, crash recovery
Web UI design	API endpoints, pages, component library
Benchmarking notes	Methodology, headline numbers, how to run
Validation test plan	Full test matrix with 100+ test cases
TLA+ correctness models	Formal verification of core invariants
Grafana dashboards	Pre-built Prometheus dashboards for monitoring

Architecture Decision Records (ADRs)

• 001: Postgres-only
• 002: BLAKE3 uniqueness
• 003: Heartbeat + deadline hybrid
• 004: PyO3 async bridge
• 005: Priority aging
• 006: AwaTransaction as narrow SQL surface
• 007: Periodic cron jobs
• 008: COPY batch ingestion
• 009: Python sync support
• 010: Per-queue rate limiting
• 011: Weighted concurrency
• 012: Split hot and deferred job storage
• 013: Durable run leases and guarded finalization
• 014: Structured progress and metadata
• 015: Builder-side post-commit lifecycle hooks
• 016: Bridge adapters for non-sqlx transactional enqueue
• 017: Python insert-only transaction bridging

License

MIT OR Apache-2.0