Enqueuing & scheduling

Calling await task.enqueue(*args, **kwargs) serializes the arguments, pushes the job onto Redis, and returns a Job handle you can use to read status and results.

job = await add.enqueue(2, 3)
print(job.id)   # a uuid hex, unless you set one yourself

Enqueuing is async because the round-trip to Redis is async — call it from within an event loop.

Enqueuing by name

The side that enqueues doesn’t have to be the side that runs. app.send(name, *args, **kwargs) dispatches by name, so a web service can put work on the queue without importing the task module — or the dependencies it pulls in:

from ardiq import Ardiq
from fastapi import FastAPI

api = FastAPI()
queue = Ardiq(redis_url="redis://localhost:6379", queue_name="example")


@api.post("/reports")
async def create_report(user_id: int):
    job = await queue.send("build_report", user_id, format="pdf")
    return {"job_id": job.id}

Nothing is checked locally: the name is resolved by the worker that picks the task up, and one it doesn’t know fails there like any other error.

Because send forwards every keyword to the task, it takes no options of its own. For those, app.ref(name) returns the same handle @app.task gives you:

await queue.ref("build_report").options(delay_ms=60_000, priority="low").enqueue(7)
await queue.ref("build_report", priority="low").enqueue(7)   # a default lane for the handle

A ref can be enqueued but not called — there is no local function behind it.

Priority does not travel with the name

Everything else on @app.task(...) — max_retries, backoff_ms, timeout — is applied by the worker, which has the registry and can look it up. Priority is the exception: it picks which stream the task goes into, so it is settled by the producer, before the payload leaves.

A task declared @app.task(priority="high") and dispatched by name lands in the app’s default_priority instead, with no warning. Pass it at the call site:

await queue.ref("build_report", priority="high").enqueue(7)

The fallback being the middle lane makes forgetting it survivable rather than disastrous, but the work still won’t be in the lane you declared for it.

Per-call options

For one-off overrides, chain .options(...) before .enqueue(...):

await add.options(priority="high", delay_ms=5000).enqueue(2, 3)

.options(...) accepts:

Option	Type	Default	Description
task_id	str | None	a uuid	Set your own job id — also used for deduplication.
priority	str | None	task’s default	Override the priority lane for this call.
delay_ms	int	0	Wait this many ms from now before the task becomes runnable.
schedule_ms	int	0	Run at this absolute epoch-ms timestamp.
expire_ms	int	0	Drop the job if it hasn’t started within this window.

Delayed tasks

Run something after a relative delay:

# fire in 30 seconds
await reminder.options(delay_ms=30_000).enqueue(user_id)

Scheduled tasks

Run something at a specific wall-clock time, using an absolute timestamp in epoch ms:

import time

run_at = int(time.time() * 1000) + 3_600_000   # one hour from now
await digest.options(schedule_ms=run_at).enqueue()

While a job is waiting in the delayed/scheduled set, its status is scheduled.

Custom ids & deduplication

Setting task_id lets you control the job id — useful to make an enqueue idempotent: the same id won’t create a duplicate job.

await sync_account.options(task_id=f"sync:{account_id}").enqueue(account_id)

Expiry

expire_ms drops a job that has been waiting too long to start — useful for work that’s worthless if it’s stale:

# if no worker picks it up within 60s, forget it
await notify.options(expire_ms=60_000).enqueue(user_id)

Priorities

Higher-priority lanes are drained first. Define the lanes (lowest-first) on the app, then target one per task or per call:

app = Ardiq(priorities=["low", "default", "high"])

await report.options(priority="low").enqueue()    # batch work
await alert.options(priority="high").enqueue()     # jump the queue

See Defining tasks for setting a task’s default lane.

Reading the result

.enqueue(...) returns immediately with a Job. To get the outcome, see Results & introspection:

job = await add.enqueue(2, 3)
result = await job.result(timeout=5)   # waits up to 5s
print(result.value)                    # 5