Python quickstart

A task is a unit of work. A flow connects tasks by passing one task's output to the next. A run is one execution of that flow with a particular set of parameters.

1. Create the workflow

Save the following as order_pipeline.py in your project. Keep the decorated functions at module level so dagy can load them when executing the graph.

from dagy import flow, task


@task
def read_orders(source: str) -> list[dict]:
    print(f"Reading orders from {source}")
    return [
        {"order_id": "ord_101", "amount": 25, "status": "paid"},
        {"order_id": "ord_102", "amount": 40, "status": "paid"},
        {"order_id": "ord_103", "amount": 10, "status": "pending"},
    ]


@task
def select_paid(orders: list[dict]) -> list[dict]:
    return [order for order in orders if order["status"] == "paid"]


@task
def summarize(orders: list[dict]) -> dict:
    summary = {
        "paid_orders": len(orders),
        "total_amount": sum(order["amount"] for order in orders),
    }
    print(summary)
    return summary


@flow(name="order_pipeline")
def order_pipeline(source: str = "sample") -> dict:
    orders = read_orders(source)
    paid = select_paid(orders)
    return summarize(paid)


if __name__ == "__main__":
    result = order_pipeline.run_local()
    print(f"Run: {result.run_id}")
    print(f"Status: {result.status}")

The flow body describes dependencies. Put network calls, file reads, and other work inside tasks: the body also executes when dagy builds a deployment.

2. Run it locally

python order_pipeline.py

The log includes the summary and a successful run status:

{'paid_orders': 2, 'total_amount': 65}
Run: order_pipeline-...
Status: SUCCEEDED

Run IDs and timestamps vary. run_local() always executes on your computer, even after you configure a hosted workspace. It returns run metadata; it does not return the summary dictionary. The summarize task's output is recorded as an artifact.

To pass another source parameter:

python -c "from order_pipeline import order_pipeline; print(order_pipeline.run_local(source='test'))"

The sample source label is only logged. Replace read_orders with your own integration when you are ready to read real data.

3. Inspect the run

Copy the run ID from the output and substitute it for RUN_ID:

dagy runs list --local
dagy runs show RUN_ID
dagy logs RUN_ID

Local logs and artifacts are stored under ~/.dagy by default. The last 20 runs are retained by the local runner. Use local configuration to change the location or artifact size limit.

For a quick assertion in application code, the eager call returns the function result:

from order_pipeline import order_pipeline

assert order_pipeline() == {"paid_orders": 2, "total_amount": 65}

An eager call executes tasks directly and does not create a tracked DAG run. Use run_local() to check orchestration, logs, retries, and dependencies. Read execution choices before switching between them.

4. Prepare for hosted execution

Connect to your workspace to sign in, select the organization, and configure the API URL. Then follow the deployment guide to package this file, choose an environment and runtime, deploy, and trigger a run.

Use the same flow and task definitions locally and remotely. This example keeps all task code in one file because the current SDK build includes the flow's source file; it does not automatically bundle an entire project or install its dependencies.

If the first run fails

Continue with workflow patterns, retries and timeouts, or the API integration guide.