SynaFlow 🌊🧠

📖 Full documentation: humansoftware.github.io/synaflow

Write plain Python functions. SynaFlow builds the pipeline for you.

SynaFlow is a lightweight, pure-Python pipeline engine that uses Type Hints to automatically wire and execute Directed Acyclic Graphs (DAGs) with lockstep streaming and optional bounded handoff.

Why the name? Synapse + Flow. Just like synapses automatically wire neurons together, SynaFlow automatically wires your functions together based on their types. "Flow" represents the lazy, streaming nature of how data moves through those connections.

Quickstart

from collections.abc import Generator, Iterator
from typing import NamedTuple
from synaflow import pipeline, step, run


class Params(NamedTuple):
    count: int


def producer(count: int) -> Generator[int, None, None]:
    yield from range(count)


def transformer(producer: Iterator[int]) -> Generator[int, None, None]:
    for val in producer:
        yield val * 10


def consumer(transformer: Iterator[int]) -> None:
    for x in transformer:
        print(f"Consumed: {x}")


p = pipeline(
    name="example",
    params=Params,
    steps=[
        step("producer", fn=producer),
        step("transformer", fn=transformer),
        step("consumer", fn=consumer),
    ],
)

run(p, Params(count=5))

Three functions, three step() calls, zero manual wiring. SynaFlow reads the type hints and wires the DAG automatically.

What makes it different

Type-hint wiring

Parameter names match producer names — SynaFlow connects them automatically. Singular/plural/suffix synonyms work too (item → items, user_list → users).

Lazy streaming with bounded handoff

SynaFlow streams lazily by default. Multiple consumers can stay lockstep, one consumer can stay lazy while another materializes, and when you need a bounded window between stages you can set max_in_flight on the producing step.

This is especially useful for I/O-bound pipelines where one step starts work and the next resolves it, such as HTTP requests, RPC calls, or object-store reads.

Static validation at build time

Type errors, missing dependencies, circular graphs, mode conflicts — all caught when pipeline(...) is called. Materialization decisions are compiled into the Dag too: mode resolution, per-dependency eager materialization, and the resolved materializer callables are frozen before run() starts. If it compiles, it's valid. No runtime surprises.

Runtime overrides on top of the compiled contract

When you need test-time swaps without patching module globals, pass ExecutionOverrides to run() or async_run() and replace only compiled runtime dependencies such as materializers or observers. Use PIPELINE_SCOPE for pipeline-level observers. The DAG shape and semantics stay fixed; only the runtime callable changes.

from synaflow import ExecutionOverrides, Observer, PIPELINE_SCOPE, Scope

overrides = ExecutionOverrides.empty(p)
sub = Scope("payments")

overrides.resources["db"] = FakeDatabase()
overrides.observers[PIPELINE_SCOPE] = [Observer(noop_metrics)]
overrides.observers[sub.scope("validate")] = [Observer(test_recorder)]
overrides.materializers[sub.scope("normalize")] = list

For included sub-pipelines, Scope(...) is the public helper for addressing compiled step keys without hardcoding "payments__validate" by hand. Declared resources={...} are production factories. They are called when a step is injected. ExecutionOverrides.resources is optional and only replaces that provider when you want a different runtime value.

See also: docs/user_docs/advanced/testability.md

Build your own runner

The DAG compiles to a deterministic JSON contract. Write custom runners or auto-generate native DAGs for Airflow, Prefect, or Dagster.

How it compares

Detailed comparisons: Hamilton · Java Streams · LINQ

Installation

pip install synaflow

Documentation

Start here: humansoftware.github.io/synaflow

License

MIT License