Lineage and dataset catalog

These features describe dataset relationships and metadata. They do not crawl every connected system, automatically discover all SQL column relationships, or enforce a contract simply because it is registered.

Declare the data your task uses

Use the Python SDK's Asset to name a dataset and attach input/output declarations to a task:

from dagy import flow, task
from dagy.lineage import Asset

@task(
    inputs=[Asset("s3://example-orders/raw/orders.json")],
    outputs=[Asset("postgres://warehouse/public.daily_orders")],
)
def summarize_orders(rows: list[dict]) -> list[dict]:
    return [{"order_count": len(rows)}]

@flow
def order_summary(rows: list[dict]):
    return summarize_orders(rows)

Declarations record intended data relationships; the task must still perform the actual reads and writes. This illustrative task transforms data passed by its caller. Add your provider client calls for real storage access.

A dataset identity is namespace plus name. For Asset("s3://example-orders/raw/orders.json"), the namespace is s3://example-orders, the name is raw/orders.json, and its key is s3://example-orders/raw/orders.json. For a non-URI dataset, use Asset("daily_orders", namespace="analytics"), whose key is analytics/daily_orders.

You can also call task.with_assets(inputs=[...], outputs=[...]) on an existing task definition. Asset declarations may include JSON-serializable OpenLineage facets. Keep identifiers stable across producers and consumers so the graph connects correctly.

Supported execution paths record task completion lineage and may register output datasets in the catalog. Visual workflows can infer dataset references for recognized nodes. Verify recorded lineage for your actual flow and runtime; missing declarations, unsupported inference, or unavailable lineage collection can leave gaps.

Explore the graph

Open Lineage in the web app, search for a dataset, and choose upstream, downstream, or both directions. API access requires runs.read in the current workspace.

curl --fail-with-body \
  "$DAGY_API_URL/lineage/assets?search=daily_orders&limit=100" \
  -H "Authorization: Bearer $DAGY_TOKEN"

This returns assets, count, and enabled. Use the returned asset key as an opaque identifier when constructing a graph URL; URL-encode the key, especially when it contains a URI.

For the simple key analytics/daily_orders:

curl --fail-with-body \
  "$DAGY_API_URL/lineage/assets/analytics%2Fdaily_orders/graph?direction=upstream&depth=3" \
  -H "Authorization: Bearer $DAGY_TOKEN"

The response contains root, nodes, edges, and enabled. Edges have from and to dataset keys. direction accepts both (default), upstream, or downstream. Depth defaults to 3 and must be 0–25. Asset listing defaults to 200 and permits up to 1,000 results.

Inspect one run's lineage

curl --fail-with-body "$DAGY_API_URL/lineage/runs/$RUN_ID" \
  -H "Authorization: Bearer $DAGY_TOKEN"

The response contains run_id, events, count, and enabled. Events use OpenLineage-style run, job, input, output, and lifecycle fields. Use this endpoint when investigating why a dataset edge appeared or why an expected edge is missing.

External OpenLineage forwarding can be configured for a Dagy deployment by its operator. There is no per-workspace API on this surface to configure a forwarding destination; coordinate it with your administrator when needed.

Register a catalog dataset

Catalog reads require nodes.read, writes require nodes.write, and deletion requires nodes.delete. These are the current permission names; there is no separate catalog.* permission family.

curl --fail-with-body "$DAGY_API_URL/catalog/datasets" \
  -H "Authorization: Bearer $DAGY_TOKEN" \
  -H 'Content-Type: application/json' \
  --data '{
    "namespace":"analytics",
    "name":"daily_orders",
    "owner":"data-platform@example.com",
    "description":"Daily order totals, ready for reporting.",
    "schema":[
      {"name":"business_date","type":"date","required":true,"nullable":false},
      {"name":"order_count","type":"integer","required":true,"nullable":false}
    ],
    "freshness_sla_seconds":86400,
    "volume_bounds":{"min_rows":1,"max_rows":1000},
    "tags":["reporting","orders"],
    "contract_version":"1.0.0"
  }'

The 201 response contains the dataset record, identified by dataset_key. Namespace defaults to default; namespace and name each permit up to 200 characters. Optional contract_ref can link the dataset to the location where your team maintains its agreement.

The schema above describes your dataset's public contract. It is not a Dagy storage schema. Catalog metadata does not automatically validate data or reject producers. Add quality checks or contract enforcement to the flow for that behavior.

Search, edit, and check freshness

List also supports owner and limit (default 200, maximum 1,000); it returns datasets and count. Search returns query, results, and count, with a default limit of 50 and maximum 200.

PUT is a partial metadata update: omitted fields are retained. For example, send {"owner":"analytics@example.com","tags":["orders","certified"]}. Deleting metadata returns 204 and does not delete data from your provider.

Dataset detail includes a contract view and a computed freshness object:

{"status":"fresh","age_seconds":3600,"sla_seconds":86400}

fresh means the most recently recorded materialization is within the configured freshness window. stale means it is outside that window. unknown means a usable freshness window or materialization timestamp is missing. A manual catalog registration alone does not establish that the dataset was produced recently. Inspect your execution path and recorded output assets when freshness stays unknown.

Troubleshoot missing metadata

Lineage capture is best effort, and catalog freshness is based on recorded materializations. Treat missing data as an observability gap to investigate, not proof that a dependency or update does not exist.