akadze

Postgres-backed task queue: enqueue work and run it in a worker. Periodic schedules run in that same process.

Shape is familiar if you know Celery. The broker is PostgreSQL (FOR UPDATE SKIP LOCKED + lease), not Redis or RabbitMQ.

Status: alpha (0.3.x). Public API may still change between minors until 1.0; after Beta the surface freezes. Bugs → Issues.

Goals

• Small library, no product domain
• Tasks, retries, lease recovery
• Periodic cron and interval schedules run inside the worker
• Clear Python call surface; CLI for migrate, worker, jobs, and queues

Non-goals (for now)

• HTTP / REST / web UI (another library or app owns that)
• Replacing any product service queue
• Multi-broker adapters
• Distributed tracing / full observability suite

First task

Install the library, point it at the application database, and create the akadze schema. The tables live in that schema; they are not part of the application's own migrations.

poetry install
export AKADZE_DATABASE_URL=postgresql://akadze:akadze@localhost:5432/akadze
akadze migrate

Save this as demo.py. enqueue runs in the caller's transaction, so a rollback removes the job.

import asyncio

from akadze import Akadze

app = Akadze(database_url="postgresql://akadze:akadze@localhost:5432/akadze")


@app.task("hello")
async def hello(name: str) -> str:
    return f"hello {name}"


async def main() -> None:
    async with app.engine.begin() as connection:
        await hello.using(session=connection).enqueue(name="ada")


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

Run the worker from the directory that contains demo.py. It loads demo:app from there, claims the job, runs hello, and sets the row to succeeded. Stop it with Ctrl-C: the worker takes no new jobs and waits up to 10 seconds (shutdown_timeout) for the ones already running. Whatever is still running goes back to the queue, and an attempt is not spent. Periodic tasks run in this same process; there is no separate beat.

If a heartbeat does not succeed within 30 seconds, the worker stops and puts unfinished jobs back on the queue. An attempt is not spent. A synchronous task runs in a thread. Stopping the worker does not stop that thread.

A claim that fills every free slot is followed by another claim immediately. Otherwise the worker waits up to poll_interval (1 second). The wait is at least half of that, so idle workers do not poll in step or spin. A job can run more than once if its worker disappears.

python demo.py
akadze worker demo:app

Split task modules with a thin include. Each module exposes register(app) and attaches @app.task / @app.periodic / app.register there — no package walk, no import-for-side-effect:

# tasks/hello.py
def register(app: Akadze) -> None:
    @app.task("hello")
    async def hello(name: str) -> str:
        return f"hello {name}"


# app.py
app = Akadze(database_url="postgresql://akadze:akadze@localhost:5432/akadze")
app.include("tasks.hello")

When dependencies exist before the live Akadze instance, declare a plain callable and bind it later. Canonical DI path: factory builds the app → include modules → app.register unbound callables (closures over deps) → optional add_schedules. No global app bind and no Celery shared_task.

# tasks/hello.py — no app, no decorator
async def hello(name: str, *, runner: Runner) -> str:
    return await runner.greet(name)


# bootstrap after assemble_runtime / DI
def create_app(deps: Deps) -> Akadze:
    app = Akadze(database_url="postgresql://akadze:akadze@localhost:5432/akadze")
    app.include("tasks.ops")

    async def _hello(name: str) -> str:
        return await hello(name, runner=deps.runner)

    app.register(_hello, name="hello")  # same options as @app.task
    # app.add_schedules([...])  # optional, after tasks exist
    return app

app.register(fn, *, name, queue=..., max_attempts=..., timeout=..., priority=...) is equivalent to @app.task(...) without decorating on a live instance. It returns the registered Task.

Schedules can also come from app config without a @periodic registrar per service. Parse YAML/TOML/JSON in the app, turn rows into already-parsed specs (cron text or timedelta — not duration strings), then call add_schedule / add_schedules. The task must already be in app.tasks; the worker tick is unchanged.

from datetime import timedelta

@app.task("ops.prune", queue="ops")
async def prune() -> None:
    ...

# after include / task registration — values already parsed by the app
app.add_schedule("ops.prune", every=timedelta(seconds=30), overlap=False)
# or many: app.add_schedules([{"name": "ops.prune", "cron": "0 * * * *"}, ...])

akadze.jobs.state is then succeeded. A higher priority is claimed first. result is optional. Pass enqueue options (delay, run_at, priority, unique_key, limit_key, expires_at) through using(...), not as task arguments. expires_at must be timezone-aware. A job that has already expired is cancelled before it runs. A second enqueue with the same active unique_key raises DuplicateJob and does not abort the rest of the caller's transaction.

Enqueue the follow-up job on the complete_tx connection so it commits only if this job succeeds:

from akadze import current

async with current().complete_tx() as connection:
    await next_step.using(session=connection).enqueue()

To commit your own writes together with the transition to succeeded, take the connection from current().complete_tx(). If this run no longer owns the job, or the job was cancelled or has expired, the block raises and those writes roll back. An error inside the block rolls the writes back and spends an attempt.

from akadze import current

async with current().complete_tx() as connection:
    await connection.execute(...)

akadze.testing.drain(app) runs every job that is ready now. assert_enqueued(app, "hello", name="ada") checks that a queued or running job has those arguments.

queue_snapshot(engine) reads committed rows: how many jobs are in each queue and state, how long the oldest ready job has waited, and how many are running. list_jobs(engine, queue=..., state=..., limit=50) returns JobSummary rows (no args/result/errors/meta), ordered by priority, then run_at, then id. akadze jobs list prints one summary per line. list_workers(engine) returns WorkerSummary rows from akadze.workers (id, hostname, pid, queues, started_at, heartbeat_at), ordered by heartbeat_at desc, then id. akadze workers list prints one summary per line. akadze.metrics.render_metrics(engine) returns Prometheus text for those counts, lag, busy, and akadze_queue_paused (not exported from the package root; no HTTP). akadze metrics (or akadze metrics render) prints that text to stdout. Count outcomes in an on_transition hook. Do not log task arguments or results.

requeue(session, job_id) returns a failed job to queued in the caller's transaction. It keeps attempt, errors, and args, and raises max_attempts when needed. request_cancel(session, job_id) sets cancel_requested_at on a queued or running job; other states and missing ids are a no-op. The worker applies the cancellation. pause_queue / resume_queue stop and restore claims for one queue; jobs already running finish. CLI: akadze jobs requeue JOB_ID, akadze jobs cancel JOB_ID, akadze queues pause NAME, akadze queues resume NAME.

set_limit(session, key, max_running=..., max_starts=..., window=...) writes policies for a limit_key in akadze.limits. Pass the key through using(..., limit_key=...); limits never live on the job row. max_running caps how many may be running cluster-wide. max_starts + window cap how many successful claims may start in that window (tracked in akadze.limit_starts, not by scanning jobs). One call may set concurrency, rate, or both; omitted fields keep their previous values. list_limits(engine) returns LimitSummary rows (key, max_running, max_starts, window). clear_limit(session, key) deletes that limits row (idempotent) and makes the key unlimited — a non-empty limit_key without a row is claimed like a job with no limit. Empty/None limit_key has no limit. When a limits row exists, concurrency and rate checks still apply. This is separate from process-local Worker.slots. CLI: akadze limits list, akadze limits clear KEY.

Do not copy this package's SQL into the application's Alembic history. Deploy runs akadze migrate, or await migrate(database_url), against the same database. The engine disables asyncpg's statement cache so a transaction-mode pool such as PgBouncer can sit in front.

Tests use AKADZE_DATABASE_URL and drop schema akadze in that database. The database name must be akadze or end with _test.

Celery migration

Primitives map to familiar Celery habits; there is no Redis/Rabbit broker and no canvas/chord.

Enqueue options go through using(...), not as task kwargs. No .get() / result backend; count outcomes in hooks or ops queries.

Release

Create a tag vX.Y.Z that matches the version in pyproject.toml and push it. GitHub Actions runs the tests, publishes the wheel to PyPI, and opens a GitHub release.

PyPI trusted publishing, set once on the project akadze:

• Owner: g4st3r-d3v
• Repository: akadze
• Workflow: release.yml
• Environment: pypi

After that publish succeeds, an application installs the package from PyPI:

pip install akadze

Layout

akadze/   # library package
tests/

Design notes: g4st3r-d3v/akadze-harness (HUB.md). This repo is code-only.

License

MIT