dagflows

Python SDK for authoring and running Dagflows workflows.

Installation

pip install dagflows

Requires Python 3.11+.

To enable the optional async HTTP client backend:

pip install dagflows[httpx]

Quickstart

Define a workflow and its node functions in a module:

# app/workflow.py
from dagflows import Workflow

wf = Workflow("order_pipeline", max_concurrent_nodes=4)


@wf.node()
def fetch_orders(ctx, inputs):
    return {"orders": [{"id": 1, "amount": 100}, {"id": 2, "amount": 250}]}


@wf.node(depends=[fetch_orders])
def calculate_totals(ctx, inputs):
    orders = inputs["fetch_orders"].value()["orders"]
    total = sum(order["amount"] for order in orders)
    return {"total": total}

Handler Signatures

Node functions can be synchronous or asynchronous, and may take fewer arguments than are offered. Arguments are passed positionally, so the first parameter is always ctx whatever you name it:

# Full signature
@wf.node()
def step_a(ctx, inputs):
    ...

# Context only: one parameter receives ctx, never inputs
@wf.node()
def step_c(ctx):
    ...

# No arguments
@wf.node()
def step_d():
    return {"status": "ready"}

# Async handler
@wf.node(depends=[step_a])
async def step_async(ctx, inputs):
    async for item in inputs["step_a"]:
        ...
    return {"done": True}

To reach inputs, declare both parameters. def step(inputs) receives ctx under the name inputs, which fails later and confusingly.

Node Configuration

Resource Limits and Retries

Configure execution limits and retry policies on @wf.node():

from dagflows import ExecutionConfig, RetryConfig, Workflow

wf = Workflow("data_pipeline")


@wf.node(
    execution=ExecutionConfig(timeout=300, memory_limit_mb=512),
    retry=RetryConfig(max_attempts=3, initial_backoff_ms=1000),
)
def extract_data(ctx, inputs):
    ...

Cross-Project Dependencies

To depend on a node from another project or workflow, use wf.external_node:

@wf.node(depends=[fetch_orders, wf.external_node("inventory_check")])
def fulfill_order(ctx, inputs):
    ...

Standalone Node Scripts (Optional)

In addition to defining nodes within a workflow module, you can also write standalone script files using dagflows.run:

# app/nodes/custom_task.py
import dagflows


def handler(ctx, inputs):
    data = inputs.one().value()
    return {"processed": True, "count": len(data)}


if __name__ == "__main__":
    dagflows.run(handler)

Working with Inputs

Inputs are accessed through input handles:

upstream = inputs["fetch_orders"]

# Retrieve complete value, refused if it exceeds the node's memory limit
data = upstream.value()

# Iterate without holding it all (sync and async iteration both work)
for record in upstream:
    process_record(record)

# Raw chunks for opaque content: a generator, not a bytes object
for chunk in upstream.bytes():
    sink.write(chunk)

Iteration yields one record at a time for row oriented payloads, which is what lets a node read more than it can hold. A JSON input has no records, so it yields the whole document as a single item.

content type	for x in handle yields
application/x-ndjson, text/csv	one row at a time
application/json	the whole document, once

Iteration re-opens the source each pass, so looping twice costs a second read rather than silently yielding nothing the second time.

If a node has exactly one parent dependency, use inputs.one():

single_input = inputs.one().value()

Producing Outputs

Return a Python dictionary/value directly, or use Result for explicit routing and formatting:

from dagflows import ContentType, Result

# Direct value return
return {"count": 42}

# Route downstream execution to a specific branch
return Result(output={"status": "approved"}, next="process_payment")

# Halt execution of this branch
return Result(output={}, stop=True)

# Explicit content type specification
return Result(output=stream_generator(), content_type=ContentType.NDJSON)

Error Handling

Raise Fail to signal structured execution errors with retry instructions:

from dagflows import EXECUTION, Fail

raise Fail("Payment gateway unavailable", category=EXECUTION, retry_after=30)

Available error categories:

• EXECUTION
• INFRASTRUCTURE
• TIMEOUT
• PERMANENT

Command Line Interface

The SDK provides CLI commands invoked via python -m dagflows:

Manifest Management

# Generate manifest file
python -m dagflows build manifest app.workflow -o dagflows-manifest.json

# Check if an existing manifest is up to date
python -m dagflows build manifest app.workflow --check

# Validate workflow definition without writing output
python -m dagflows build validate app.workflow

Local Development & Testing

dev run executes a node with no VM, no platform and no network. It takes a script path, so it applies to standalone node scripts:

python -m dagflows dev run app/nodes/custom_task.py --input fetch_orders=orders.json

dev fixture writes a starting input envelope, so nobody has to invent one. It takes either form:

# a standalone script
python -m dagflows dev fixture app/nodes/custom_task.py -o fixture.json

# a node declared with @wf.node
python -m dagflows dev fixture app.workflow:calculate_totals --input fetch_orders=orders.json -o m.json

It prints the command that runs the node against what it wrote, which for a decorated node goes through the interpreter because a module:function entrypoint is not a file:

DAGFLOWS_INPUT=m.json DAGFLOWS_OUTPUT=out.json python -m dagflows invoke --node calculate_totals

That is the local loop for decorator-defined nodes, since dev run needs a file to execute.

dev run accepts several options worth knowing:

--input users=rows.ndjson     a file, content type inferred from the suffix
--input users='{"n": 1}'      inline json
--memory-limit-mb 512         what the node believes it has
--inline-max-bytes 262144     when an output would offload
--keep-fixture <path>         write the envelope instead of discarding it

All CLI commands support --json for machine-readable output. Exit codes are 0 success, 1 the operation failed, 2 the command was wrong.

License

Apache-2.0