Data quality and contracts

Registering a catalog entry documents that agreement. Add enforcement to the flow to make violations affect execution.

Add a Data Quality node

In the Flow Builder, add Data Quality after the node producing rows. Connect its data_in input and choose checks such as:

{
  "checks":[
    {"kind":"row_count","min":1,"severity":"fail"},
    {"kind":"not_null","columns":["order_id"],"severity":"fail"},
    {"kind":"unique","columns":["order_id"],"severity":"fail"},
    {"kind":"values_in","column":"status","allowed":["paid","pending"],"severity":"warn"}
  ],
  "on_failure":"route"
}

This is the node configuration object. The node accepts rows represented as supported lists, JSON payloads, or dataframes.

With on_failure: "route" (default), failed data goes to fail_out and the node itself succeeds. Set on_failure: "error" to fail node execution on a blocking violation. warn records a failure without blocking. quarantine is a severity and routing signal; it does not automatically move data into separate storage. Connect an explicit storage or review step if you need that behavior.

Supported checks

The freshness check uses the newest timestamp, not every row's age. If every row must be recent, add an explicit transformation/check for that requirement. GET /quality/catalog returns the supported check catalog for your deployment.

Validate Python task output

Use check_output inside the task decorator:

from dagy import task
from dagy.expectations import check_output

@task
@check_output([
    {"kind":"row_count","params":{"min":1},"severity":"fail"},
    {"kind":"not_null","params":{"columns":["order_id"]},"severity":"fail"},
])
def extract_orders() -> list[dict]:
    return [{"order_id":"order-1042","amount":42.0}]

The decorator evaluates the returned payload. Blocking failures raise DataQualityError; warnings allow the task to continue. There is no @task(expectations=...) argument.

For explicit checks within ordinary Python code:

from dagy.expectations import expect

report = (
    expect([{"order_id":"order-1042"}])
    .row_count(min=1)
    .not_null(["order_id"])
    .unique(["order_id"])
    .evaluate()
)
for result in report.results:
    print(result.to_dict())

See the SDK reference for the complete authoring surface and local validation examples.

Enforce a data contract

Contracts support field names and types, required/nullability flags, ownership, description, version, freshness settings, and volume bounds. Load a contract from a dictionary or YAML and apply it to task output:

from dagy.contracts import contract_task, load_contract

orders_contract = load_contract({
    "name":"orders",
    "namespace":"analytics",
    "version":"1.0.0",
    "schema":[
        {"name":"order_id","type":"string","required":True},
        {"name":"amount","type":"number","required":True},
    ],
    "volume_bounds":{"min_rows":1},
})

@contract_task(orders_contract, severity="error")
def contract_checked_orders():
    return [{"order_id":"order-1042","amount":42.0}]

contract_task already creates a Dagy task; do not add a second @task wrapper. Use severity="warn" to log violations or "error" to raise ContractViolationError. There is no @task(contract=...) or .with_contract(...) interface.

Contract field validation supports string, integer, float, number, boolean, timestamp, date, object, array, and any. Timestamp/date types accept strings without proving they parse as dates; add an explicit format check when that matters.

The default payload validator checks field content on the first 1,000 rows while volume bounds consider the full payload. Freshness enforcement requires an explicit produced_at value with enforce_contract; the task decorator does not supply that timestamp. Do not assume a contract's freshness setting alone enforces task output freshness.

Use diff_contracts in your development/CI workflow to detect incompatible changes before deployment. Review producer and consumer code together when removing fields, adding required fields, or changing types; updating only catalog metadata does not migrate data or consumers.

Inspect recorded results

The web app's Settings → Data controls shows configured quality checks and flow pass rates. All quality API reads require runs.read:

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

curl --fail-with-body \
  "$DAGY_API_URL/quality/summary?flow_name=process-orders&window=7d" \
  -H "Authorization: Bearer $DAGY_TOKEN"

Run results contain results, summary, and ok. Each recorded check includes its name, severity, observed and expected values, result, and message. Summary windows use m, h, d, or w, for example 30m, 24h, 7d, or 2w.

GET /quality/checks returns supported checks and checks discovered in workspace flow definitions. Discovery is bounded to the first 200 flow records. Flow summary reads are bounded to 5,000 recorded checks; they are not an unlimited historical export.

Handle failures and missing results

Persist critical quality reports as part of the workflow when you need a durable application-level record independent of the observability surface.