Pipeline Registry & CLI¶

Once you have more than a couple of pipelines in a project, two questions start to dominate the day:

Where do they live? A file, a directory, a package?
How do I run one from the shell without writing a driver script every time?

PipelineRegistry answers the first. The synaflow CLI answers the second.

The PipelineRegistry class¶

PipelineRegistry is a validated, name-keyed mapping of (name → PipelineDef, Dag). Adding a pipeline compiles its Dag immediately, so an imported catalog never contains an invalid pipeline. Adding a root also registers every pipeline reachable through include().

from typing import NamedTuple

from synaflow import PipelineRegistry, pipeline, step


class HelloParams(NamedTuple):
    x: int = 0


def hello(x: int) -> int:
    return x


hello_pipeline = pipeline(
    name="hello",
    params=HelloParams,
    steps=[step("hello", fn=hello)],
)
catalog = PipelineRegistry()
catalog.add(hello_pipeline)

After that:

dag = catalog.get_dag("hello")   # already compiled during catalog.add(...)
dag2 = catalog.get_dag("hello")  # same compiled Dag

Note

catalog[name] gives you the PipelineDef. catalog.get_dag(name) gives you the compiled Dag. A registered definition must not be mutated. catalog.add(...) is idempotent for the same instance and rejects a different instance with the same name.

Loading a catalog module¶

PipelineRegistry.from_module("myproject.pipelines") imports that module and returns its catalog attribute. The module must expose catalog = PipelineRegistry() at the top level.

This is the convention the CLI uses. Put your pipelines and their explicit catalog.add(...) calls in a single module:

myproject/pipelines.py

from typing import NamedTuple
from synaflow import PipelineRegistry, pipeline, step


class DailyParams(NamedTuple):
    x: int = 0


def greet(x: int) -> None:
    print(f"hello, x = {x}")


_greet = pipeline(
    name="greet",
    params=DailyParams,
    steps=[step("greet", fn=greet)],
    exports="status/loaded.json",
)

catalog = PipelineRegistry()
catalog.add(_greet)

The `synaflow` CLI¶

If you ran uv add synaflow (or pip install synaflow), the install also gives you a synaflow console script. It is a thin adapter over the public Python API; you can do everything it does from python -m synaflow.

In both cases you point at a catalog module with --catalog:

synaflow --catalog myproject.pipelines list
synaflow --catalog myproject.pipelines info greet
synaflow --catalog myproject.pipelines dag greet
synaflow --catalog myproject.pipelines run greet

The CLI dispatches sync vs async automatically based on the compiled Dag.requires_async_runner flag — you don't pick the engine.

Projects with a fixed catalog can expose the same commands without requiring users to repeat --catalog:

from synaflow import SynaflowCli
from myproject.pipelines import catalog

if __name__ == "__main__":
    raise SystemExit(SynaflowCli(catalog=catalog).main())

`list` — what's registered?¶

$ synaflow --catalog myproject.pipelines list
greet   1 steps

--json emits JSON instead:

$ synaflow --catalog myproject.pipelines list --json
[
  {
    "name": "greet",
    "steps": 1
  }
]

`info` — the declared shape (no compilation)¶

$ synaflow --catalog myproject.pipelines info greet
name: greet
params: DailyParams
exports: status/loaded.json
steps (1): greet

info only reads the PipelineDef; it never calls build_dag. The steps (1) count is the declared step count, before any sub-pipeline includes are expanded.

--json returns the same fields as a JSON object — useful for tooling.

`dag` — the compiled Dag, as JSON¶

$ synaflow --catalog myproject.pipelines dag greet
{
  "name": "greet",
  "steps": [
    {
      "name": "greet",
      "fn": "greet",
      ...
    }
  ]
}

This does compile the Dag and prints whatever shape Dag.to_dict() produces. Use it for debugging or for piping the compiled structure into another tool.

`run` — execute a pipeline¶

# All default params (works when every field has a default).
synaflow --catalog myproject.pipelines run greet

# From a JSON file (object with one key per params field).
synaflow --catalog myproject.pipelines run greet --params-file params.json

# Direct flags are generated from the pipeline's Params fields.
# --x overrides values from --params-file.
synaflow --catalog myproject.pipelines run greet --x 99
synaflow --catalog myproject.pipelines run greet --params-file p.json --x 99

Simple params fields become kebab-case flags: initial_date becomes --initial-date. The supported direct types are str, int, float, bool, bytes (base64) and lists of those values. Booleans use paired flags, such as --dry-run and --no-dry-run; lists are repeatable.

Complex values — mappings, sets, tuples and nested params objects — belong in --params-file as JSON. Direct flags override values from that file. --param key=value was removed; use the corresponding typed flag instead.

Add --no-observers to run to disable every pipeline and step observer for that invocation while keeping normal materializers and resources.

Unknown fields and missing required fields are reported as synaflow: Unknown params field(s) for DailyParams: [...] etc. — not as a stack trace.

Lifecycle hooks¶

The catalog-mode synaflow console script doesn't expose hooks — it is the same CLI for everyone. Project-specific entry points built on SynaflowCli can plug two callables around each run invocation:

from synaflow import SynaflowCli, PreRunContext, PostRunContext
from myproject.pipelines import catalog

def audit(context: PostRunContext) -> None:
    # Always runs after `run` finishes — on success and on failure.
    log.info("pipeline %s %s", context.pipeline.name, context.outcome.status)

def enrich_tenant(context: PreRunContext):
    # Runs after CLI parsing, before the pipeline starts.
    # Return value replaces the params the executor sees.
    return context.params._replace(tenant=os.environ["MYPROJECT_TENANT"])

if __name__ == "__main__":
    raise SystemExit(
        SynaflowCli(
            catalog=catalog,
            pre_run=enrich_tenant,
            post_run=audit,
        ).main()
    )

PreRunContext carries the resolved pipeline, dag, params, and observers_enabled. pre_run must return a value of the same type as the pipeline's params — anything else raises TypeError and the pipeline never starts.

PostRunContext adds outcome: RunOutcome, where outcome.status is either "succeeded" or "failed" and outcome.error is the original exception on failure (or None). The pipeline exception is re-raised after post_run runs; if post_run itself raises, its exception is re-raised with the original chained as __cause__.

When to use hooks vs observers

Hooks wrap the whole run (one shot, per CLI invocation). Observers fire inside the pipeline on per-step lifecycle events (one per step, per run). Reach for hooks for boundary effects — audit, metrics emission, Slack alerts — and observers for fine-grained progress signals.

End-to-end: a real catalog in 30 seconds¶

myproject/pipelines.py

from typing import NamedTuple
from synaflow import PipelineRegistry, pipeline, step


class DailyParams(NamedTuple):
    x: int = 0
    label: str = "default"


def ingest(x: int) -> int:
    print(f"ingest: x = {x}")
    return x


def transform(x: int, label: str) -> dict:
    return {"label": label, "value": x * 2}


_ingest = pipeline(
    name="daily_ingest",
    params=DailyParams,
    steps=[step("ingest", fn=ingest), step("transform", fn=transform)],
    exports="status/loaded.json",
)


catalog = PipelineRegistry()
catalog.add(_ingest)

$ synaflow --catalog myproject.pipelines list
daily_ingest    2 steps

$ synaflow --catalog myproject.pipelines info daily_ingest
name: daily_ingest
params: DailyParams
exports: status/loaded.json
steps (2): ingest, transform

$ synaflow --catalog myproject.pipelines run daily_ingest \
    --x 7 --label '"first run"'
ingest: x = 7
$ echo $?
0

The catalog module is the single source of truth — Python code, the CLI, and any future automation read from the same PipelineRegistry.

Pipeline Registry & CLI¶

The PipelineRegistry class¶

Loading a catalog module¶

The synaflow CLI¶

list — what's registered?¶

info — the declared shape (no compilation)¶

dag — the compiled Dag, as JSON¶

run — execute a pipeline¶