JobResult

Handle to an enqueued job. Provides methods to check status and retrieve results, both synchronously and asynchronously.

Returned by task.delay(), task.apply_async(), queue.enqueue(), and canvas operations.

Properties

job.id

job.id -> str

The unique job ID (UUIDv7, time-ordered).

job.status

job.status -> str

Current job status. Fetches fresh from the database on every access.

Returns one of: "pending", "running", "complete", "failed", "dead", "cancelled".

job = add.delay(2, 3)
print(job.status)  # "pending"
# ... after worker processes it ...
print(job.status)  # "complete"

A note on naming. The per-job status is "complete" (singular). The aggregate count bucket from queue.stats() is "completed" (past tense). Status filters such as list_jobs(status=…) accept either spelling.

job.progress

job.progress -> int | None

Current progress (0–100) if reported by the task via current_job.update_progress(). Returns None if no progress has been reported. Refreshes from the database.

job.error

job.error -> str | None

Error message if the job failed. None if the job hasn't failed. Refreshes from the database.

job.errors

job.errors -> list[dict]

Full error history for this job — one entry per failed attempt. Each dict contains:

job = flaky_task.delay()
# ... after retries ...
for err in job.errors:
    print(f"Attempt {err['attempt']}: {err['error']}")

job.dependencies

job.dependencies -> list[str]

List of job IDs this job depends on. Returns an empty list if the job has no dependencies. See Dependencies.

job.dependents

job.dependents -> list[str]

List of job IDs that depend on this job. Returns an empty list if no other jobs depend on this one.

Both are direct edges only, one hop in each direction, and both query storage on every read rather than caching — a property in a loop is a query in a loop. Declared as get_dependencies and get_dependents; for the transitive picture, queue.job_dag(job_id) walks both directions at once.

Edges never cross namespaces — an enqueue that would create one is rejected — so the lists are scoped by the anchor job. A job in another namespace reports no edges, exactly like an unknown id.

Methods

job.to_dict()

job.to_dict() -> dict

Return all job fields as a plain dictionary. Useful for JSON serialization (e.g. in the dashboard or FastAPI integration). Refreshes from storage first, so the snapshot is current rather than whatever this handle last saw.

It deliberately omits the stored result: that is the encoded blob the declared result_bytes carries, and JSON-serializing it would be lossy or invalid depending on the serializer. Use result(), which decodes it. payload_bytes is the same story on the way in — the encoded arguments, before the serializer has turned them back into Python values. Both are PyJob properties; reach for them only when you need the wire bytes, such as re-encoding a payload under a different serializer.

job.result()

job.result(
    timeout: float = 30.0,
    poll_interval: float = 0.05,
    max_poll_interval: float = 0.5,
) -> Any

Block until the job completes and return the deserialized result. Uses exponential backoff for polling — starts at poll_interval and gradually increases to max_poll_interval.

Raises:

• TimeoutError — if the job doesn't complete within timeout
• TaskFailedError — if the job status is "failed"
• MaxRetriesExceededError — if the job status is "dead" (all retries exhausted)
• TaskCancelledError — if the job status is "cancelled"
• SerializationError — if result deserialization fails

from flexiq import TaskFailedError, MaxRetriesExceededError, TaskCancelledError

job = add.delay(2, 3)
result = job.result(timeout=10)  # blocks, returns 5

# Handle specific failure modes
try:
    result = job.result(timeout=10)
except TaskCancelledError:
    print("Job was cancelled")
except MaxRetriesExceededError:
    print("Job exhausted all retries")
except TaskFailedError:
    print("Job failed")

await job.aresult()

await job.aresult(
    timeout: float = 30.0,
    poll_interval: float = 0.05,
    max_poll_interval: float = 0.5,
) -> Any

Async version of result(). Uses asyncio.sleep() instead of time.sleep(), so it won't block the event loop.

Raises: same as result().

job = add.delay(2, 3)
result = await job.aresult(timeout=10)

job.stream()

job.stream(
    timeout: float = 60.0,
    poll_interval: float = 0.5,
) -> Iterator[Any]

Iterate over partial results published by the task via current_job.publish(). Yields each result as it arrives, stops when the job reaches a terminal state.

job = batch_process.delay(items)
for partial in job.stream(timeout=120):
    print(f"Got: {partial}")

await job.astream()

async for partial in job.astream(
    timeout: float = 60.0,
    poll_interval: float = 0.5,
) -> AsyncIterator[Any]

Async version of stream(). Uses asyncio.sleep so it won't block the event loop.

async for partial in job.astream(timeout=120):
    print(f"Got: {partial}")