Job Context

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

Usage

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.

Properties

current_job.id

current_job.id -> str

The unique ID of the currently executing job.

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

current_job.task_name

current_job.task_name -> str

The registered name of the currently executing task.

current_job.retry_count

current_job.retry_count -> int

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

@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

current_job.queue_name -> str

The name of the queue this job is running on.

Methods

current_job.update_progress()

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

@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:

job = process_files.delay(files)

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

current_job.publish()

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.

@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:

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

How it works

Sync tasks (thread pool):

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

Async tasks (native async pool):

1. Before execution, set_async_context() sets a contextvars.ContextVar token
2. current_job checks contextvars first; if a token is set it returns that context
3. 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.