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.

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).

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}")