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

A handler declares what it needs, so a node that only reads its parents does not carry an unused ctx. Ask for arguments by name, in any order:

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

# Inputs only
@wf.node(depends=[step_a])
def step_b(inputs):
    ...

# Context only
@wf.node()
def step_c(ctx):
    ...

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

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

Annotating

Annotate for editor completion, which the SDK ships py.typed to support:

from dagflows import Ctx, Inputs


@wf.node()
def step(ctx: Ctx, inputs: Inputs):
    inputs.one().value()      # completion on both

An annotation also frees the name, so call the parameters whatever suits you:

@wf.node()
def step(payload: Inputs, context: Ctx):
    ...

A parameter that is neither named nor annotated is refused, naming the fix:

transform: cannot tell what to pass to 'data'.
Name it 'ctx' or 'inputs', or annotate it:

    def transform(data: Inputs): ...

That refusal happens while the manifest is emitted, so a bad signature fails the build rather than the first run.

Node Configuration

Resource Limits and Retries

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

A policy on the workflow is the default every node inherits. A node stating its own overrides it field by field, so the settings it leaves out still apply:

from dagflows import ExecutionConfig, RetryCategory, RetryConfig, Workflow

wf = Workflow(
    "data_pipeline",
    retry=RetryConfig(max_attempts=3, initial_backoff_ms=1000, max_backoff_ms=30_000),
)


@wf.node(
    execution=ExecutionConfig(timeout=300, memory_limit_mb=512),
    # Overrides workflow retry defaults field by field
    retry=RetryConfig(
        max_attempts=5,
        retry_on=[RetryCategory.INFRASTRUCTURE, RetryCategory.EXECUTION],
    ),
)
def extract_data(ctx, inputs):
    ...


@wf.node(
    depends=[extract_data],
    # Setting max_attempts=1 opts out of retries
    retry=RetryConfig(max_attempts=1),
)
def charge_card(ctx, inputs):
    ...

max_attempts counts the first run, so 1 means run once and do not retry. Backoff doubles per attempt up to max_backoff_ms.

Leaving a setting out is not the same as setting it to zero: unstated means the platform decides, which is what lets a workflow default reach a node at all.

By default only the platform's own failures are retried. retry_on widens that, and RetryCategory names what can be asked for, so a typo fails where it was written rather than at deploy:

permanent has no member on purpose: the platform reports it when running the node again cannot change the outcome, so asking to retry it is refused.

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.

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