Job Context

Per-job context for the currently executing task — id, retry count, progress, publish.

Per-job context for the currently executing task. Provides access to job metadata and controls from inside a running task.

Usage

python

from flexiq.context import current_job

# or directly:
from flexiq import current_job

current_job is a module-level singleton. It works in both sync and async tasks:

Sync tasks — reads from threading.local, isolated per worker thread.
Async tasks — reads from a contextvars.ContextVar, isolated per concurrent coroutine even when multiple async tasks run on the same event loop.

current_job can only be used inside a running task. Accessing it outside a task raises RuntimeError.

Properties

`current_job.id`

python

current_job.id -> str

The unique ID of the currently executing job.

python

@queue.task()
def process(data):
    print(f"Running as job {current_job.id}")
    ...

`current_job.task_name`

python

current_job.task_name -> str

The registered name of the currently executing task.

`current_job.retry_count`

python

current_job.retry_count -> int

How many times this job has been retried. 0 on the first attempt.

python

@queue.task(max_retries=3)
def flaky_task():
    if current_job.retry_count > 0:
        print(f"Retry attempt #{current_job.retry_count}")
    call_external_api()

`current_job.queue_name`

python

current_job.queue_name -> str

The name of the queue this job is running on.

Methods

`current_job.update_progress()`

python

current_job.update_progress(progress: int) -> None

Update the job's progress percentage (0–100). The value is written directly to the database and can be read via job.progress or queue.get_job().

python

@queue.task()
def process_files(file_list):
    for i, path in enumerate(file_list):
        handle(path)
        current_job.update_progress(int((i + 1) / len(file_list) * 100))

Read progress from the caller:

python

job = process_files.delay(files)

# Poll progress
import time
while job.status == "running":
    print(f"Progress: {job.progress}%")
    time.sleep(1)

`current_job.publish()`

python

current_job.publish(data: Any) -> None

Publish a partial result visible to job.stream() consumers. Use this to stream intermediate data from long-running tasks.

data must be JSON-serializable. It is stored as a task log entry with level LogLevel.RESULT, distinguishing it from regular logs.

python

@queue.task()
def process_batch(items):
    for i, item in enumerate(items):
        result = process(item)
        current_job.publish({"item_id": item.id, "status": "ok"})
        current_job.update_progress(int((i + 1) / len(items) * 100))
    return {"total": len(items)}

Consumer side:

python

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

`current_job.step`

python

current_job.step -> StepContext

The durable step recorder for this attempt. Built lazily and cached on the context, so a task that never takes a step pays nothing.

Member	Description
`run(name, fn, *, key=None)`	Run `fn` once per run and memoize its result. A replay after a crash, a retry or a sleep returns the recorded value without calling `fn` again. `arun()` is the await twin; it commits off the event loop so a slow write does not stall every other coroutine on the worker.
`sleep(duration, *, name=None, key=None)`	Sleep, ending the attempt if the deadline is ahead: the claim is released and the job returns to `pending` until then, so a sleeping job holds no worker slot and cannot be timed out while it waits. Costs no retry — the retry count, retry budget, circuit breaker and metrics are all untouched. `asleep()` is the await twin.
`sleep_until(when, *, name=None, key=None)`	The absolute-instant form; reach for it when the deadline means something outside the job (a billing date, a market open), because it is unaffected by how many times the attempt replayed. `asleep_until()` is the await twin.
`idempotency_key`	The key to hand a downstream service for the step running now. Stable across a retry, a sleep/wake and an operator's DLQ retry, and no serializer or codec touches it. Readable only inside a step body — outside one there is no step for it to name.
`run_key`	The id this durable run began under: the job's own id, except across a DLQ retry, which mints a new job for the same run and keeps the original key so a charge is not made twice.

The sleep deadline is fixed by the first commit. Replaying a "1h" sleep wakes at the original instant rather than an hour later each time, which is what stops a crash loop from producing a sleep that outlives the job. Name your sleeps: a sequence reading sleep#0, sleep#1, sleep#2 tells nobody which one diverged.

Steps need a worker or an attached executor to commit through. A task running outside both raises StepUnavailableError as a control signal the body cannot catch away, rather than running the step un-memoized. queue.test_mode() is the exception: the step runs and is not memoized, but the idempotency key is still derived, so a test can assert on what the step would have sent.

Declared names

Declared	What it is
`supports_steps`	Whether the backend has a step store at all — a capability probe, not the gate; a session refuses on its own when the store is missing.
`open_step_session`	Opens the attempt's session, on `WorkerSteps` (a worker) or `AttachedSteps` (an executor). The handle rides with the dispatch, so the fence names this run — `WorkerSteps.owner` is the worker that won the claim, never the `Queue` handle, because one process may run several workers off one queue.
`inherited_worker_steps`	How a prefork child picks up the parent's step handle after the fork.
`run_key`	Read back by `step.run_key`.
`sleep_for` / `sleep_until`	Behind `step.sleep()` / `step.sleep_until()`; `sleep_for` takes `duration_ms`, while the API accepts a duration string.
`StepDecision.step_key` / `.memoized`	One `run()`'s verdict: the step's identity, and the recorded bytes when it is a memo hit (`None` on a first run).
`StepSleepOutcome.step_key` / `.wake_at` / `.elapsed`	One sleep's verdict: its identity, the fixed wake instant (Unix ms), and whether the deadline had already passed — `elapsed=True` is what lets a replayed sleep return immediately instead of ending the attempt again.

How it works

Sync tasks (thread pool):

Before execution, the Rust worker calls _set_context() with the job's metadata
current_job reads from threading.local — each worker thread has independent storage
After the task completes (success or failure), _clear_context() resets the thread-local

Async tasks (native async pool):

Before execution, set_async_context() sets a contextvars.ContextVar token
current_job checks contextvars first; if a token is set it returns that context
After the coroutine finishes, clear_async_context() resets the token

This means concurrent async tasks on the same event loop each see their own isolated context — there is no cross-task interference.