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.authoring 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.runtime 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):
    ...

Typed nodes

Any parameter that is neither ctx nor inputs is the node's input: the one parent's output, decoded into whatever the parameter is annotated with. The return annotation says what the node produces. Nothing is declared twice, and both ends are reflected into the manifest so the platform can check every edge before a run starts:

# app/typed.py
from collections.abc import Iterator
from dataclasses import dataclass

from dagflows.authoring import Workflow
from dagflows.runtime import Input

wf = Workflow("typed")


@dataclass
class Order:
    id: int
    amount: int


@dataclass
class Orders:
    orders: list[Order]


@dataclass
class Totals:
    total: int


@dataclass
class Line:
    id: int
    cents: int


@wf.node()
def fetch_orders() -> Orders:
    return Orders([Order(1, 100), Order(2, 250)])


@wf.node(depends=[fetch_orders])
def calculate_totals(orders: Orders) -> Totals:
    return Totals(sum(o.amount for o in orders.orders))


@wf.node(depends=[fetch_orders])
def export_lines(orders: Orders) -> Iterator[Line]:
    for o in orders.orders:
        yield Line(o.id, o.amount * 100)


@wf.node(depends=[export_lines])
def count_lines(lines: Iterator[Line]) -> int:
    return sum(1 for _ in lines)


@wf.node(depends=[fetch_orders])
def peek(orders: Input[Orders]) -> dict:
    return {"bytes": orders.size, "first": orders.value().orders[0].id}


@wf.node(depends=[fetch_orders, calculate_totals])
def report(inputs) -> dict:
    orders = inputs[fetch_orders].value()      # Orders, typed by the handle
    totals = inputs[calculate_totals].value()  # Totals
    return {"orders": len(orders.orders), "total": totals.total}

The annotation on the input parameter decides the form it takes:

A node with several parents takes inputs and reaches each parent through its handle, which types it; inputs["fetch_orders"] still works and is untyped.

Decoding accepts dataclasses, TypedDict, NamedTuple, pydantic models, Enum, Literal, Optional, unions, list/dict/set/tuple, datetime/date, Decimal and bytes (base64 on the wire). A value that does not fit fails the run naming the field, input.orders[0]: Order needs 'amount', which the input does not have, rather than as an AttributeError three calls later. A handler without annotations is never refused; its types are simply "anything".

The manifest carries what each node expects and produces, as a JSON Schema subset the builder checks along every edge, including edges into nodes written in Go or TypeScript:

"io": {
  "inputs": {"fetch_orders": {"shape": "value", "schema": {"type": "object", "title": "Orders", "...": "..."}}},
  "output": {"shape": "value", "content_type": "application/json", "schema": {"type": "object", "title": "Totals", "...": "..."}}
}

Two signatures are refused while the manifest is emitted, so they fail the build rather than the first run: two parameters that would both be the input (def h(data, more)), and an input parameter on a node that does not have exactly one parent, each with the fix in the message.

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.authoring import Execution, RetryCategory, Retry, Workflow

wf = Workflow(
    "data_pipeline",
    retry=Retry(max_attempts=3, initial_backoff_ms=1000, max_backoff_ms=30_000),
    execution=Execution(machine="gp-2"),
)


@wf.node(
    # Overrides workflow execution defaults with a named machine tier
    execution=Execution(machine="gp-8", timeout_secs=300),
    # Overrides workflow retry defaults field by field
    retry=Retry(
        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=Retry(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.runtime


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


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

Working with Inputs

Inputs are accessed through input handles:

upstream = inputs["fetch_orders"]

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

# Retrieve complete value, for an input small enough to hold
data = upstream.value()

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

Reach for iteration first. It reads an inline handful of rows and a stored file of any size through the same loop, so a node written this way keeps working when its parent grows. value() is the convenience for an input you know is small: it holds the whole payload in memory and is refused when it exceeds the node's memory limit.

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.

Reach a parent through the handle @wf.node returned instead of its key and the input is typed by what that node declared it produces: inputs[fetch_orders].value() is an Orders when fetch_orders is annotated -> Orders, and iterating inputs[export_lines] yields Line records when it is annotated -> Iterator[Line].

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.runtime import ContentType, Result

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

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

# Or to several; children not named are skipped. Handles work as well as keys.
return Result(output={"status": "approved"}, next=["process_payment", "notify"])

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

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

# Metadata stored with the run, for whoever reads it later
return Result(output={"status": "approved"}, meta={"reviewer": "ana"})

next absent means every child runs; a list runs exactly the children it names, in graph order, and skips the rest; stop=True skips them all. Result[T] is generic, so a handler annotated -> Result[Totals] produces a Totals as far as its children and the manifest are concerned.

Generators

A handler may be a generator itself, which is the shortest way to write a node whose output does not fit in memory. async def generators work the same way:

# app/generating.py
from dagflows.authoring import Transfer, Workflow

wf = Workflow("generating")


@wf.node()
def enrich(inputs):
    for order in inputs.one():
        yield {**order, "tax": order["amount"] * 0.2}


@wf.node()
async def enrich_async(inputs):
    async for order in inputs.one():
        yield {**order, "tax": order["amount"] * 0.2}

Three rules apply:

• yield is one row. The output is NDJSON, even where a single object is yielded once. Return the object instead when a node produces one value.
• A generator node is atomic. A failure at row 900 replays from row 0; rows already sent are discarded, never resumed.
• return inside a generator is dropped. Only yielded rows reach the output.

Laziness on its own does not bound what the node holds. Declare Transfer(max_output_mb=...) for that: it is what asks the platform for a multipart upload, which is what lets parts leave as they fill instead of the whole output being buffered for a single request.

# app/bounded.py
from dagflows.authoring import Transfer, Workflow

wf = Workflow("bounded")


@wf.node(transfer=Transfer(max_output_mb=4096))
def export_all(inputs):
    yield from inputs.one()

Without it a large output is refused rather than truncated, naming max_output_mb as the remedy.

Error Handling

Raise Fail to signal structured execution errors with retry instructions:

from dagflows.failure import EXECUTION, Fail

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

# A stable machine readable name and any JSON that explains it travel with the message
raise Fail("card declined", category=EXECUTION, code="card_declined", details={"last4": "4242"})

Available error categories:

• EXECUTION
• INFRASTRUCTURE
• TIMEOUT
• PERMANENT

Long-Running Nodes

A node whose handler takes rows in and yields rows out is written the same way whether it runs once or stays alive. mode=Mode.STREAM asks the platform to keep it running and feed it live, with backpressure and cancellation (ctx.done), on a platform that offers channels; on one that does not, it runs exactly as a Mode.ONCE node would, so the code is the same either way:

# app/streaming.py
from collections.abc import AsyncIterator
from dataclasses import dataclass

from dagflows.authoring import Mode, Workflow

wf = Workflow("streaming")


@dataclass
class Tick:
    symbol: str
    price: float


@dataclass
class Enriched:
    symbol: str
    price: float
    doubled: float


@wf.node()
async def ticks() -> AsyncIterator[Tick]:
    for price in (1.0, 2.0):
        yield Tick("ABC", price)


@wf.node(depends=[ticks], mode=Mode.STREAM)
async def enrich(quotes: AsyncIterator[Tick]) -> AsyncIterator[Enriched]:
    async for tick in quotes:
        yield Enriched(tick.symbol, tick.price, tick.price * 2)

Event Triggers

A trigger is a typed source in the graph. The event is the input of every node that depends on it, and its delivery metadata is ctx.trigger, so business data and platform metadata never mix:

# app/events.py
from dataclasses import dataclass

from dagflows.authoring import Workflow
from dagflows.runtime import Ctx

wf = Workflow("events")


@dataclass
class OrderPlaced:
    order_id: str
    amount: int


placed = wf.trigger("order_placed", event=OrderPlaced)


@wf.node(depends=[placed])
def on_order(event: OrderPlaced, ctx: Ctx) -> dict:
    ctx.log.info("delivery %s", ctx.trigger.id if ctx.trigger else "local")
    return {"received": event.order_id}

The manifest lists the trigger with its schema and the node as triggered_by it; which kinds of trigger exist and how they are bound is the platform's side.

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