Sub-Workflows & Scheduling

Nest workflows for composition, and schedule workflows on a cron.

Sub-workflows

Use WorkflowProxy.as_step() to embed one workflow inside another:

@queue.workflow("etl")
def etl_pipeline(region):
    wf = Workflow()
    wf.step("extract", extract, args=(region,))
    wf.step("load", load, after="extract")
    return wf

@queue.workflow("daily")
def daily_pipeline():
    wf = Workflow()
    wf.step("eu_etl", etl_pipeline.as_step(region="eu"))
    wf.step("us_etl", etl_pipeline.as_step(region="us"))
    wf.step("reconcile", reconcile, after=["eu_etl", "us_etl"])
    return wf

run = daily_pipeline.submit()

How it works

1. Parent workflow submits the child workflow via queue.submit_workflow() with parent_run_id
2. The child runs independently with its own nodes and status
3. When the child completes/fails, the tracker updates the parent node
4. Downstream steps in the parent evaluate normally

Cancellation cascade

Cancelling the parent cascades to all active child workflows:

run.cancel()  # Cancels parent + all child sub-workflows

Failure

If a child workflow fails at runtime, the parent node is marked FAILED. Downstream steps follow the parent's on_failure strategy.

The same holds for failures at submission time — if the child's factory raises or the DAG fails to compile when the parent node becomes evaluable, the parent node is marked FAILED immediately (rather than leaving the outer run hanging), and the parent run finalizes normally.

Cron-scheduled workflows

Stack @queue.periodic() on top of @queue.workflow():

flexiq cron is 6-field, seconds first (sec min hour day month weekday), so 0 0 2 * * * is 02:00:00 daily — not Celery's 5-field crontab. Prepend a seconds field when porting a Celery schedule.

@queue.periodic(cron="0 0 2 * * *")    # sec=0 min=0 hour=2 → 2:00 AM daily
@queue.workflow("nightly_analytics")
def nightly():
    wf = Workflow()
    wf.step("extract", extract_clickstream)
    wf.step("aggregate", build_dashboards, after="extract")
    return wf

Each cron trigger submits a new workflow run. Under the hood, a bridge task _wf_launcher_nightly_analytics is registered that calls proxy.submit().