Python API¶
Everything below is importable from akadze unless the heading says otherwise. Signatures come from the code on main.
App and tasks¶
akadze.Akadze ¶
Akadze(
*,
engine: AsyncEngine | None = None,
database_url: str | None = None,
pool_size: int | None = None,
grace: timedelta = timedelta(minutes=1),
retention: timedelta = timedelta(days=7),
)
The application: an engine, the registered tasks and the schedules.
Pass database_url and the app builds an async engine it owns (and
closes in aclose), with asyncpg's statement cache off. Pass engine
instead to share a pool; the app then leaves it open. One of the two is
required.
pool_size sets the pool of the engine the app builds (SQLAlchemy's
default is 5, plus 10 overflow). A worker needs at least
slots x queues + 2 connections. With your own engine, size its
pool there instead.
grace is how late a periodic fire may still be enqueued (default one
minute). retention is how long finished jobs and periodic runs are
kept before the worker deletes them (default seven days). Keep
retention longer than grace: a periodic run deleted inside the
grace window can be enqueued again.
Example
app = Akadze(database_url="postgresql://localhost/app", pool_size=25)
@app.task("mail.send", queue="mail", max_attempts=6)
async def send_mail(user_id: int) -> None: ...
engine
instance-attribute
¶
engine = engine or create_async_engine(
async_database_url(database_url or ""),
connect_args={"statement_cache_size": 0},
**sizing,
)
include ¶
include(*modules: str) -> None
Import each module and call its register(app).
Modules own registration (@app.task / @app.periodic /
app.register). There is no package walk and no
import-for-side-effect convention.
register ¶
register(
fn: Callable[P, Any],
*,
name: str,
queue: str = "default",
max_attempts: int = 3,
timeout: timedelta | None = None,
priority: int = 0,
backoff: BackoffPolicy | None = None,
) -> Task[P]
Bind a callable into the task registry without decorating on app.
Same options and validation as @app.task. Returns the registered
Task so the caller can enqueue it.
timeout is not a hard limit for a sync callable: it runs in a
thread, which keeps running after the timeout while the job is retried
or failed, so a retry can overlap it, and the thread stays out of
Worker.slots. Registering one with timeout logs a warning on
logger akadze.app.
task ¶
task(
name: str,
*,
queue: str = "default",
max_attempts: int = 3,
timeout: timedelta | None = None,
priority: int = 0,
backoff: BackoffPolicy | None = None,
) -> Callable[[Callable[P, Any]], Task[P]]
Register the decorated function as task name and return its Task.
queue is where its jobs go; workers claim from the queues they are
started with. max_attempts is the budget including the first run.
A run longer than timeout counts as an error and is retried.
priority is the default for its jobs (higher is claimed first).
backoff maps the attempt number to the delay before the next one
(default: exponential with jitter, capped at 300 s). Arguments are
validated by a model built from the signature. Same options as
register.
periodic ¶
periodic(
name: str,
*,
cron: str | None = None,
every: timedelta | None = None,
overlap: bool = True,
queue: str = "default",
max_attempts: int = 3,
timeout: timedelta | None = None,
timezone: str | None = None,
backoff: BackoffPolicy | None = None,
) -> Callable[[Callable[..., Any]], Task[Any]]
Register the decorated function as a task that runs on a schedule.
Give exactly one of cron (five fields, UTC unless timezone
names an IANA zone) and every (a timedelta). Every worker runs
the schedule; each due fire is enqueued once across the cluster.
overlap=False holds a fire back while the previous job is queued or
running, for at most the app's grace. queue, max_attempts,
timeout and backoff are those of task; periodic jobs have
priority 0.
add_schedule ¶
add_schedule(
name: str,
*,
cron: str | None = None,
every: timedelta | None = None,
overlap: bool = True,
queue: str | None = None,
timezone: str | None = None,
) -> None
Attach a periodic schedule to an already registered task.
Same cron/every/overlap/queue/timezone semantics as @periodic.
queue defaults to the task's queue. Returns nothing.
add_schedules ¶
add_schedules(
specs: Sequence[ScheduleSpec | Mapping[str, object]],
) -> None
Attach many schedules from already-parsed specs. Returns nothing.
akadze.task.Task ¶
Task(
app: Any,
name: str,
fn: Callable[P, Any],
*,
queue: str,
max_attempts: int,
timeout: timedelta | None,
priority: int,
backoff: BackoffPolicy | None = None,
)
Bases: Generic[P]
A registered task. @app.task and app.register return one.
Enqueue it through using: await task.using(session=conn).enqueue(...).
using ¶
using(
*,
session: AsyncConnection | AsyncSession,
delay: timedelta | None = None,
run_at: datetime | None = None,
priority: int | None = None,
max_attempts: int | None = None,
unique_key: str | None = None,
limit_key: str | None = None,
expires_at: datetime | None = None,
meta: Mapping[str, Any] | None = None,
) -> BoundTask[P]
Bind enqueue options; the jobs are inserted on session, in its transaction.
delay or run_at (timezone-aware) set the earliest start.
priority and max_attempts override the task's defaults for
these jobs. While a job with unique_key is queued or running,
another enqueue with that key raises DuplicateJob; keys are shared
across all tasks and queues. limit_key
names the limits that apply (see set_limit). A job not started by
expires_at is cancelled. meta is a small JSON object with
string keys, at most 8 KiB, for filtering and hooks; not for arguments
or secrets.
akadze.task.BoundTask ¶
BoundTask(
task: Task[P],
*,
session: AsyncConnection | AsyncSession,
delay: timedelta | None,
run_at: datetime | None,
priority: int,
max_attempts: int,
unique_key: str | None,
limit_key: str | None,
expires_at: datetime | None,
meta: Mapping[str, Any] | None = None,
)
Bases: Generic[P]
A task bound to a session and enqueue options; Task.using returns one.
enqueue
async
¶
enqueue(*args: args, **kwargs: kwargs) -> None
Insert one job with these arguments (by name). Returns nothing.
Raises EnqueueError on positional or invalid arguments and invalid
options, and DuplicateJob when unique_key is taken; neither
inserts a row.
DuplicateJob does not abort the caller's transaction.
enqueue_many
async
¶
enqueue_many(items: Iterable[Mapping[str, Any]]) -> None
Insert one job per item (arguments by name) with the shared options.
All items are validated before anything is inserted. unique_key
applies to one job, so it is rejected here.
akadze.context.Context ¶
Context(run: _Run)
Handle for the job executing on this asyncio task.
complete_tx
async
¶
complete_tx() -> AsyncIterator[AsyncConnection]
Yield a connection. On a clean exit the job is finished in that transaction.
Worker¶
akadze.Worker ¶
Worker(
app: Any,
*,
queues: list[str] | None = None,
slots: int = 1,
heartbeat_interval: timedelta = timedelta(seconds=5),
heartbeat_ttl: timedelta = timedelta(seconds=30),
poll_interval: timedelta = timedelta(seconds=1),
shutdown_timeout: timedelta = timedelta(seconds=10),
fair: bool = False,
)
Claims jobs from queues and runs them; also runs schedules and maintenance.
slots is how many jobs run at once per queue. Liveness is a heartbeat
every heartbeat_interval; a worker whose heartbeat is older than
heartbeat_ttl is presumed dead: another worker requeues its jobs,
spending an attempt with the default backoff (failed on the last one).
An idle worker polls about every poll_interval; a worker whose slots
are all taken claims again as soon as one of those runs ends. On stop it
waits up to shutdown_timeout for running jobs, then returns the rest to
the queue without spending an attempt. fair=True shares slots across
limit_key groups within a priority instead of taking the oldest jobs.
akadze worker module:app builds one and calls serve.
serve
async
¶
serve(stop: Event) -> None
Run until stop is set or no heartbeat succeeds within heartbeat_ttl.
The heartbeat runs as its own task, so a slow or hung step cannot delay it. When it goes stale, the loop is cancelled and running jobs are stopped before other workers rescue them. In the loop, a transient database error (connection lost, restart, deadlock) is logged and retried with backoff; any other error stops the worker.
Outcomes¶
akadze.Retry ¶
Retry(delay: timedelta | None = None)
Bases: Exception
Run the job again later. This spends an attempt.
akadze.Snooze ¶
Snooze(delay: timedelta)
Bases: Exception
Run the job again later. This does not spend an attempt.
BackoffPolicy is Callable[[int], timedelta]: it takes the number of the attempt that just failed and returns the delay before the next one. It must be a plain function, or an object whose __call__ is not async.
Errors¶
akadze.AkadzeError ¶
Bases: Exception
The caller used akadze incorrectly.
akadze.EnqueueError ¶
Bases: AkadzeError
The job was not inserted.
akadze.DuplicateJob ¶
Bases: EnqueueError
An active job already uses this unique_key.
Jobs¶
akadze.request_cancel
async
¶
request_cancel(
session: AsyncConnection | AsyncSession, job_id: int
) -> None
Ask a queued or running job to stop. The worker applies the cancellation.
A queued job is cancelled without running. A running job is not interrupted:
if its body returns a storable result, or its complete_tx block
commits, it ends succeeded; otherwise cancelled.
akadze.requeue
async
¶
requeue(
session: AsyncConnection | AsyncSession, job_id: int
) -> None
Return a failed job to the queue. Keeps attempt, errors, and args.
akadze.list_jobs
async
¶
list_jobs(
engine: AsyncEngine,
*,
queue: str | None = None,
state: str | None = None,
limit: int = _LIMIT_DEFAULT,
meta: Mapping[str, Any] | None = None,
) -> tuple[JobSummary, ...]
Return job summaries. Optional filters; does not mutate rows.
meta keeps jobs whose metadata contains every given key/value.
akadze.JobSummary
dataclass
¶
JobSummary(
id: int,
task: str,
queue: str,
state: str,
priority: int,
attempt: int,
max_attempts: int,
run_at: datetime,
started_at: datetime | None,
finished_at: datetime | None,
meta: Mapping[str, Any] = (
lambda: MappingProxyType({})
)(),
)
meta
class-attribute
instance-attribute
¶
meta: Mapping[str, Any] = field(
default_factory=lambda: MappingProxyType({}), hash=False
)
akadze.queue_snapshot
async
¶
queue_snapshot(engine: AsyncEngine) -> QueueSnapshot
Return depth, lag of the oldest ready job, and how many jobs are running.
akadze.QueueSnapshot
dataclass
¶
QueueSnapshot(
counts: tuple[StateCount, ...],
lag: timedelta | None,
busy: int,
)
akadze.StateCount
dataclass
¶
StateCount(queue: str, state: str, jobs: int)
akadze.job.Job
dataclass
¶
Job(
id: int,
task: str,
queue: str,
priority: int,
state: str,
args: dict[str, Any],
run_count: int,
attempt: int,
max_attempts: int,
snoozes: int,
unique_key: str | None,
limit_key: str | None,
worker_id: UUID | None,
errors: list[dict[str, str]],
result: Any,
cancel_requested_at: datetime | None,
expires_at: datetime | None,
expired: bool = False,
meta: dict[str, Any] = dict(),
)
Queues and limits¶
akadze.pause_queue
async
¶
pause_queue(
session: AsyncConnection | AsyncSession, queue: str
) -> None
Stop new claims from this queue. Running jobs finish. Idempotent.
akadze.resume_queue
async
¶
resume_queue(
session: AsyncConnection | AsyncSession, queue: str
) -> None
Allow claims from this queue again. Idempotent if the queue was not paused.
akadze.set_limit
async
¶
set_limit(
session: AsyncConnection | AsyncSession,
key: str,
max_running: int | None = None,
*,
max_starts: int | None = None,
window: timedelta | None = None,
) -> None
Upsert limit policies for a limit_key. Omitted fields keep previous values.
akadze.clear_limit
async
¶
clear_limit(
session: AsyncConnection | AsyncSession, key: str
) -> None
Delete the limits row and the start marks for key. Idempotent if the key has no row.
Removes the policy so the key is unlimited: jobs that still carry this limit_key are claimed like jobs with no limit. A later set_limit on the key starts its rate window from zero.
akadze.list_limits
async
¶
list_limits(
engine: AsyncEngine,
) -> tuple[LimitSummary, ...]
Return every limits row. Does not mutate state.
akadze.LimitSummary
dataclass
¶
LimitSummary(
key: str,
max_running: int | None,
max_starts: int | None,
window: timedelta | None,
)
Workers and schema¶
akadze.list_workers
async
¶
list_workers(
engine: AsyncEngine,
) -> tuple[WorkerSummary, ...]
Return registered workers. Ordered by heartbeat_at desc, then id. No mutation.
akadze.WorkerSummary
dataclass
¶
WorkerSummary(
id: UUID,
hostname: str,
pid: int,
queues: tuple[str, ...],
started_at: datetime,
heartbeat_at: datetime,
)
akadze.migrate
async
¶
migrate(database_url: str) -> None
Apply pending migrations. This command does not return database state.
akadze.applied_versions
async
¶
applied_versions(database_url: str) -> list[str]
Return applied migration ids. This query does not write.
Hooks¶
akadze.hooks.Hooks ¶
Hooks()
Lists of callables, sync or async, that app.hooks holds.
on_enqueue(connection, job) runs in the enqueue transaction.
before_run(job) runs before the task body and after_run(job)
after it returns, inside its timeout; an error there counts as a failed
run. on_transition(connection, job, from_state, to_state) runs in the
transaction in which enqueue, a worker or a rescue changes the job's
state; requeue does not call it. It must not raise: an error rolls the
transition back, and the job stays running on its worker. Append to the
lists.
Metrics (akadze.metrics)¶
akadze.metrics.render_metrics
async
¶
render_metrics(engine: AsyncEngine) -> str
Return Prometheus exposition text for jobs, lag, busy, and pauses.
Testing (akadze.testing)¶
akadze.testing.drain
async
¶
drain(app: Akadze, *, limit: int = _DRAIN_LIMIT) -> None
Run every job that is ready now. Jobs scheduled in the future stay queued.
Ready means a claim can take it, or a step cancels it (cancel requested,
expired). Jobs of a paused queue stay queued. Limits are not checked: a job
held back by max_starts keeps drain stepping until its window passes or
limit runs out.
Raises AkadzeError if a ready job's task is not registered on app (a
worker would leave it queued), or if jobs are still ready after limit runs.
akadze.testing.assert_enqueued
async
¶
assert_enqueued(
app: Akadze, task: str, **arguments: Any
) -> None
Raise AssertionError unless a queued or running job has this task and these arguments.
Arguments are compared in their stored JSON form, so models, UUID and
datetime can be passed as the task receives them.