Data Lineage & Provenance
Dagy captures **data-level lineage** (which datasets a run read and wrote), in addition to task-level dependencies, and can emit it in **OpenLineage** format.
Declaring assets (SDK)
Flow authors declare the datasets a task reads/writes using
dagy.lineage.Asset and task.with_assets(...) (see src/dagy/lineage.py).
Assets have a namespace and name and can carry facets; a URI like
s3://bucket/prefix is parsed into a namespace + name.
For builtin nodes, infer_assets_from_node((node_type, config)) maps a node's
type and config to its input/output assets automatically, so lineage works
without manual annotation for supported node types.
Capture & emission (API)
The lineage package (src/dagy_api/lineage/) provides:
capture_task_lineage(...)— the executor integration hook; call it after a task reaches a terminal state.build_run_event/validate_run_event— construct/validate an OpenLineage 1.xRunEvent.emit_openlineage_event(...)— build, persist, and optionally forward.
Events are stored in the table named by DAGY_LINEAGE_TABLE and, if
DAGY_OPENLINEAGE_URL is set, POSTed to that external OpenLineage endpoint
fire-and-forget (lineage never blocks or fails the run). Every read/write is
org-scoped. When DAGY_LINEAGE_TABLE is unset the store is a no-op, so importing
the package is always safe.
Query API
| Method & path | Purpose |
|---|---|
GET /lineage/assets | list known assets for the org |
GET /lineage/assets/{asset_key}/graph | upstream/downstream graph for an asset |
GET /lineage/runs/{run_id} | lineage events for a run |
The graph traversal powers impact analysis ("what breaks if this table changes?").