Metadata-Version: 2.5
Name: dagflows
Version: 0.5.0
Summary: SDK for authoring and running Dagflows workflows
Project-URL: Homepage, https://github.com/dagflows/sdk-python
Project-URL: Repository, https://github.com/dagflows/sdk-python
Project-URL: Issues, https://github.com/dagflows/sdk-python/issues
Author: dagflows
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: dag,orchestration,workflow
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.11
Provides-Extra: httpx
Requires-Dist: httpx>=0.27; extra == 'httpx'
Description-Content-Type: text/markdown

# dagflows

Python SDK for authoring and running Dagflows workflows.

## Installation

<!-- not-tested: installs from the package index -->
```bash
pip install dagflows
```

Requires Python 3.11+.

To enable the optional async HTTP client backend:

<!-- not-tested: installs from the package index -->
```bash
pip install dagflows[httpx]
```

## Quickstart

Define a workflow and its node functions in a module:

```python
# 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:

```python
# 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:

```python
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:

```python
@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:

```python
# 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:

| annotation | the parameter receives |
| --- | --- |
| a type such as `Orders` | the value, decoded into it |
| `Iterator[Line]` or `AsyncIterator[Line]` | the parent's records one at a time, each decoded, however large the parent is |
| `Input[Orders]` | the lazy handle: `.size`, `.bytes()`, `.value()` |
| `bytes` | the parent's bytes, whole |
| none | the value as plain JSON |

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:

```json
"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:

```python
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:

| category | what it means |
| --- | --- |
| `INFRASTRUCTURE` | the platform could not run the node |
| `TIMEOUT` | the node ran out of its time budget |
| `EXECUTION` | the node itself failed |

`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`:

```python
@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`:

```python
# 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:

```python
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.

| 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.

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()`:

```python
single_input = inputs.one().value()
```

## Producing Outputs

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

```python
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:

```python
# 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.

```python
# 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:

```python
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:

```python
# 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:

```python
# 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

```bash
# 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:

```bash
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:

```bash
# 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:

```bash
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:

```text
--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
