Examples and recipes

Start with the order pipeline quickstart if you are new to flows and tasks.

Parallel branches

Use independent branches when two operations can share an input without waiting for one another. A downstream task waits for both outputs.

Save as parallel_report.py:

from dagy import flow, task


@task
def load_orders() -> list[dict]:
    return [{"amount": 25}, {"amount": 40}]


@task
def count_orders(rows: list[dict]) -> int:
    return len(rows)


@task
def sum_orders(rows: list[dict]) -> int:
    return sum(row["amount"] for row in rows)


@task
def report(count: int, amount: int) -> dict:
    result = {"count": count, "amount": amount}
    print(result)
    return result


@flow
def parallel_report() -> dict:
    rows = load_orders()
    return report(count_orders(rows), sum_orders(rows))


if __name__ == "__main__":
    print(parallel_report.run_local(max_workers=2).status)

python parallel_report.py

The report contains {"count": 2, "amount": 65} and the run succeeds. max_workers controls local concurrency; setting concurrency_limit on a task does not impose a local per-task limit. Avoid mutating shared input objects in concurrent tasks.

Dynamic mapping

Use .map() when an upstream step returns a collection whose size is known only at runtime.

Save as mapped_orders.py:

from dagy import flow, task


@task
def order_ids() -> list[str]:
    return ["ord_101", "ord_102", "ord_103"]


@task
def fetch_order(order_id: str, currency: str) -> dict:
    return {"order_id": order_id, "currency": currency}


@task
def summarize_orders(orders: list[dict]) -> dict:
    result = {"count": len(orders), "first": orders[0]["order_id"]}
    print(result)
    return result


@flow
def mapped_orders() -> dict:
    orders = fetch_order.map(order_ids(), currency="USD")
    return summarize_orders(orders)


if __name__ == "__main__":
    print(mapped_orders.run_local().status)

The mapped output is an ordered list, so the result has count=3 and first="ord_101". Local mapped items execute sequentially. Mapping creates separate logical work items; it does not promise parallel execution on every runtime.

To pair multiple input collections, replace the .map() call with:

orders = fetch_order.expand(
    order_id=["ord_101", "ord_102"],
    currency=["USD", "EUR"],
)

This creates two calls, one for each pair. Unequal collections stop at the shortest input. With continue_on_error=True, a DAG map attempts remaining items after a failure but the mapped node still fails overall; it is not a partial-success result contract.

Conditional branches

Use a DAG condition to choose tasks based on a value produced by an upstream task. Save as route_order.py:

from dagy import branch, flow, task, when


@task
def amount() -> int:
    return 120


@task
def manual_review(value: int) -> str:
    print(f"Review required for {value}")
    return "review"


@task
def automatic_processing(value: int) -> str:
    return "automatic"


@flow
def route_order() -> None:
    value = amount()
    branch(
        when(value) > 100,
        if_true=lambda: manual_review(value),
        if_false=lambda: automatic_processing(value),
    )


if __name__ == "__main__":
    print(route_order.run_local().status)

manual_review runs and automatic_processing is skipped. This example routes work; it does not create a hosted human-approval request. See approvals and operational controls for workspace review actions.

branch() records graph conditions and returns None. Run this example with run_local() or deploy it; calling route_order() eagerly raises an error. when supports truthiness, .is_falsy(), .eq(value), .ne(value), comparisons, .in_(values), and .not_in(values).

Compose child flows

Call a decorated child flow from its parent to reuse a group of tasks:

from dagy import flow, task


@task
def normalize(value: str) -> str:
    return value.strip().lower()


@flow
def prepare_email(email: str) -> str:
    return normalize(email)


@task
def contact(email: str) -> dict:
    return {"email": email}


@flow
def prepare_contact(email: str = "Person@Example.com") -> dict:
    return contact(prepare_email(email))


if __name__ == "__main__":
    print(prepare_contact.run_local().status)

The child receives resolved inputs at execution time. Keep child functions importable and available in the target deployment. The local runner caps nested flow depth at 32. Python child flows are distinct from the preview visual-builder Subflow node; consult its catalog entry.

Validate data before publishing

Use expectations to verify that records are suitable for a downstream system. Put @task above @check_output so the task wrapper calls the validator.

from dagy import check_output, flow, task


@task
@check_output([
    {"kind": "row_count", "min": 1},
    {"kind": "not_null", "columns": ["order_id"]},
    {"kind": "unique", "columns": ["order_id"]},
    {"kind": "values_in", "column": "status", "allowed": ["paid", "pending"]},
])
def checked_orders() -> list[dict]:
    return [{"order_id": "ord_101", "status": "paid"}]


@task
def prepare_export(rows: list[dict]) -> dict:
    return {"ready_to_export": len(rows)}


@flow
def validated_export() -> dict:
    return prepare_export(checked_orders())


if __name__ == "__main__":
    print(validated_export.run_local().status)

A blocking check raises DataQualityError; downstream export work is skipped. Add "severity": "warn" to a check to record a warning without failing. quarantine is also blocking; the Python decorator does not move records into a quarantine destination automatically.

For direct validation, use the fluent builder:

from dagy import expect

rows = [{"order_id": "ord_101", "amount": 25}]
report = (
    expect(rows)
    .row_count(min=1, max=1000)
    .not_null(["order_id"])
    .unique(["order_id"])
    .distribution("amount", min=0, max=10000)
    .evaluate()
)
report.raise_for_status()
print(report.to_dict())

Additional checks are .values_in(column, allowed), .matches(column, pattern, full_match=False), and .freshness(column, max_age_seconds). Freshness measures the newest parseable timestamp; it does not require every row to be recent. Reports expose ok, results, failures, warnings, quarantined, and a pass_rate between 0 and 1. See data quality for hosted reporting and routing.

Enforce a data contract

A contract defines the data your producer promises to consumers. Import contract helpers from dagy.contracts. contract_task already creates a task, so an additional @task is unnecessary.

from dagy import flow
from dagy.contracts import DataContract, FieldSpec, VolumeBounds, contract_task

orders_contract = DataContract(
    dataset="orders",
    namespace="commerce",
    owner="data-team@example.com",
    version="1",
    schema=(
        FieldSpec("order_id", "string"),
        FieldSpec("amount", "number"),
    ),
    volume_bounds=VolumeBounds(min_rows=1, max_rows=1000),
)


@contract_task(orders_contract)
def export_orders() -> list[dict]:
    return [{"order_id": "ord_101", "amount": 25}]


@flow
def contracted_export() -> list[dict]:
    return export_orders()


if __name__ == "__main__":
    print(contracted_export.run_local().status)

Use severity="warn" on contract_task to log violations without raising. The default error severity raises ContractViolationError with .violations. Supported types are string, integer, float, number, boolean, timestamp, date, object, array, and any; numeric fields reject booleans. timestamp and date accept strings without validating their format.

load_contract(mapping) reads a public contract dictionary. load_contracts_file(path) accepts one YAML contract, a list, or a contracts: list. diff_contracts(old, new) reports field removals, type changes, and required/nullability tightening as breaking changes.

The default validator checks schema on the first 1,000 rows and volume on the whole collection. To validate every row or apply a freshness SLA, call the validator explicitly:

from datetime import datetime, timezone
from dagy.contracts import enforce_contract

rows = [{"order_id": "ord_101", "amount": 25}]
enforce_contract(
    orders_contract,
    rows,
    sample_limit=len(rows),
    check_extra_fields=True,
    produced_at=datetime.now(timezone.utc),
)

Freshness enforcement needs both freshness_sla_seconds on the contract and produced_at on validation. The decorator does not supply a production timestamp. There is no @task(contract=...) argument or .with_contract() method in this SDK. Read catalog and contracts for managing published contracts.

Declare data lineage

Attach the identities of datasets you read and write to the task that performs the operation:

from dagy import Asset, task


@task(
    inputs=[Asset("s3://example-source/orders.json")],
    outputs=[Asset("daily_orders", namespace="commerce")],
)
def transform_orders(rows: list[dict]) -> list[dict]:
    return [row for row in rows if row.get("status") == "paid"]

Asset takes a dataset URI or a name with an explicit namespace. Optional facets contain JSON-compatible dataset metadata. task.with_assets(inputs=[...], outputs=[...]) returns an additional declaration on a copy of an existing task. These declarations describe data movement; they do not create credentials, provision storage, or read the dataset. See lineage for inspecting relationships and exporting lineage.

Use the repository examples

A source checkout also contains examples under examples/01_basics through examples/09_complex. Some example entrypoints call .deploy() and depend on your profile. Import their flow and call .run_local() explicitly when testing locally, or use the self-contained examples/docs/order_pipeline.py quickstart.

Before moving a recipe to production, replace sample data, configure secrets and dependencies, choose an environment, set client-level timeouts, and check the current limits. Run the full workflow on representative input and inspect both status and output artifacts.