Result Streaming

Stream partial results from long-running tasks via current_job.publish() and job.stream().

Stream intermediate results from long-running tasks. Instead of waiting for the final result, consumers receive partial data as it becomes available.

Partial results are delivered by polling the database (the task_logs table), not pushed by a message broker. For streaming progress over HTTP instead of polling in Python, see the FastAPI SSE integration.

Publishing partial results

Inside a task, call current_job.publish(data) to emit a partial result:

python

from flexiq import current_job

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

publish() accepts any JSON-serializable value — dicts, lists, strings, numbers.

Consuming with `stream()`

The caller iterates over partial results as they arrive:

python

job = process_batch.delay(items)

for partial in job.stream(timeout=120, poll_interval=0.5):
    print(f"Processed item {partial['item_id']}: {partial['status']}")

# After stream ends, get the final result
final = job.result(timeout=5)

stream() polls the database for new partial results, yields each one, and stops when the job reaches a terminal state (complete, failed, dead, cancelled).

Parameter	Type	Default	Description
`timeout`	`float`	`60.0`	Maximum seconds to wait
`poll_interval`	`float`	`0.5`	Seconds between polls

Async streaming

Use astream() in async contexts:

python

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

FastAPI SSE

The built-in FastAPI progress endpoint supports streaming partial results:

code

GET /jobs/{job_id}/progress?include_results=true

Events include partial results alongside progress:

code

data: {"status": "running", "progress": 25}
data: {"status": "running", "progress": 25, "partial_result": {"item_id": 1, "status": "ok"}}
data: {"status": "running", "progress": 50}
data: {"status": "complete", "progress": 100}

Patterns

ETL pipeline

python

@queue.task()
def etl_pipeline(source_tables):
    for table in source_tables:
        rows = extract(table)
        transformed = transform(rows)
        load(transformed)
        current_job.publish({
            "table": table,
            "rows_processed": len(rows),
            "status": "loaded",
        })
    return {"tables": len(source_tables)}

ML training

python

@queue.task()
def train_model(config):
    model = build_model(config)
    for epoch in range(config["epochs"]):
        loss = train_epoch(model)
        current_job.publish({
            "epoch": epoch + 1,
            "loss": float(loss),
            "lr": optimizer.param_groups[0]["lr"],
        })
    save_model(model)
    return {"final_loss": float(loss)}

Batch processing with error tracking

python

@queue.task()
def process_orders(order_ids):
    for oid in order_ids:
        try:
            process_order(oid)
            current_job.publish({"order_id": oid, "status": "ok"})
        except Exception as e:
            current_job.publish({"order_id": oid, "status": "error", "error": str(e)})
    return {"total": len(order_ids)}

How it works

publish() stores data as a task log entry with level LogLevel.RESULT, reusing the existing task_logs table. No new tables or Rust changes are needed.

stream() polls get_task_logs(job_id), filters for entries whose level matches LogLevel.RESULT (stored as "result"), tracks the last-seen timestamp, and yields only new entries. It stops when the job's status becomes terminal.