Skip to content

v0.0.14

New: SwitchNode and LabelSwitch

Route data to branch DAGs based on a string label.

  • SwitchNode — dag.switch(branches, label) routes upstream data to one of several branch DAGs based on a LabelSwitch value. Branches can be DAGs, @dag_factory functions, or plain callables.
  • MapSwitchNode — dag.map(dag.switch([...])) combines map and switch: maps over a collection, routing each element through the switch.
  • LabelSwitch — string type alias for the routing key, extracted from upstream output as (label, data) tuple or {"__switch__": label}.

New: FlatMap

  • MapNode flatmap — when a sub-DAG returns a list, elements are flattened into the parent result list. Scalar results appended as before. Applies to list inputs only (dict inputs unaffected).

New: DAG Inspection API

Pydantic models for structured provenance queries.

  • DagRun, NodeExecution, EdgeTraversal — Pydantic models with timezone-aware UTC datetimes and nested structure.
  • DagRun.latest_update() — most recent timestamp across all records.
  • DagRun.nodes_changed_since(dt) — nodes updated after a given time.
  • get_recent_runs() / get_child_runs() return list[DagRun] with full nested data loaded in one DB cycle.
  • parent_run_id on dag_runs — sub-DAG runs link to parent. get_child_runs() uses recursive CTE for all descendants.

New: Singleton DAG

  • Namespace.get_as_dag(nsref) — returns the underlying DAG for DAG nodes or wraps plain callables in a single-node DAG. Used by TriggerManager for uniform DagRunner execution path.

Changed

  • CompositeDagNode refactored — sub_dag moved from base to MapNode/CallNode. SwitchNode has branches: dict[str, Dag]. _run_sub_dag takes Dag directly.
  • Batch provenance writes — complete_node_with_edges() and fail_node_and_finish_run() batch related DB writes. All provenance writes in DagRunner._execute() run in asyncio executor.
  • safe_json_dumps — handles Pydantic BaseModel via model_dump() in provenance serialization. Shared across dag_provenance.py, dag_runner.py, and trigger.py.

Fixes

  • 6-field cron timing — croniter with seconds-first format now uses second_at_beginning=True. Fixes ~7 minute delay for */15 * * * * *.
  • Trigger provenance — fire() uses DagRunner with the manager's provenance instead of dag() (which used NullProvenance).
  • Windows temp cleanup — ignore_cleanup_errors=True on all TemporaryDirectory in tests to avoid SQLite lock errors.

Documentation

  • Updated module docstrings for namespace, dag_runner, dag_provenance.
  • Added SwitchNode, MapSwitch, flatmap sections to composable DAGs how-to.
  • Updated index page and reference pages for new types.