Workflows
Build multi-step pipelines as directed acyclic graphs.
Build multi-step pipelines as directed acyclic graphs.
Build multi-step pipelines as directed acyclic graphs. Define steps, wire dependencies, and let flexiq handle execution order, parallelism, failure propagation, and state tracking — all backed by a Rust engine with dagron-core for graph algorithms.
from flexiq import Queue
from flexiq.workflows import Workflow
queue = Queue(db_path="tasks.db")
@queue.task()
def extract(): # fetch source data and write it to a store (DB, S3, …)
...
@queue.task()
def transform(): # read what extract wrote, clean it, write the result back
...
@queue.task()
def load(): # read the cleaned data, write it to the warehouse
...
wf = Workflow(name="etl_pipeline")
wf.step("extract", extract)
wf.step("transform", transform, after="extract")
wf.step("load", load, after="transform")
run = queue.submit_workflow(wf)
result = run.wait(timeout=60)
print(result.state) # WorkflowState.COMPLETEDafter sets execution order, not data flow: a step is not automatically
called with its predecessor's return value — each runs with the args you give it
in wf.step(...). To pass a value between steps, use fan-out / fan-in or read
WorkflowContext.results. See Building workflows.
A workflow is a graph declared up front, one job per node, distributable and
inspectable. When the sequence is just the shape of one function — charge,
wait, email — reach for
durable steps instead: same replay-safety,
written with if and for, inside a single job.
submit_workflow() enqueues the step jobs; they execute only while a worker is
consuming the queue. Start one first — flexiq worker --app tasks:queue, or
queue.run_worker() in a background thread — otherwise run.wait() blocks
until it times out. Workflows ship in the standard pip install flexiq wheel;
no extra install is needed.