Metadata-Version: 2.4
Name: pyflowstep
Version: 0.8.0
Summary: A lightweight, typed Python library for composing functions into readable, reusable flows that can be defined as JSON or TOML.
Keywords: flow,pipeline,composition,functional,fluent-interface,json,json-schema,toml,workflow,dsl
Author: aaltatan
Author-email: aaltatan <a.altatan@gmail.com>
License-Expression: GPL-3.0-only
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Dist: pyargprocessors>=0.2.0
Requires-Dist: pyvocabulary>=0.1.3
Requires-Python: >=3.12
Project-URL: Homepage, https://github.com/aaltatan/pyflowstep
Project-URL: Repository, https://github.com/aaltatan/pyflowstep
Project-URL: Issues, https://github.com/aaltatan/pyflowstep/issues
Description-Content-Type: text/markdown

# pyflowstep

A lightweight, typed Python library for composing functions into readable, reusable **flows** — a functional replacement for the fluent-interface (method-chaining) pattern, with flows that can be defined, validated and stored as **JSON**.

`pyflowstep` focuses on a functional style:

- steps are plain functions: `(subject, *args, **kwargs) -> subject`
- flows are immutable values that compose with `>>`
- step arguments are validated and processed when a flow is **built**, not halfway through running it
- raw JSON values become typed arguments with a [pyargprocessors](https://github.com/aaltatan/pyargprocessors) `Process` marker next to the parameter
- steps get external objects (a mailer, a database session) through FastAPI-style `Depends`
- the caller hands per-run values to a flow by name: `flow(page, user=user)` fills every `user: User = Input()`
- registry-based registration keeps steps organized and discoverable
- flow definitions can be compiled from dictionaries or JSON
- every registry can describe its flow language as a JSON Schema

It pairs naturally with its siblings [pyspecification](https://github.com/aaltatan/pyspecification) (predicates) and [pyformula](https://github.com/aaltatan/pyformula) (formulas). Its only dependency is [pyargprocessors](https://github.com/aaltatan/pyargprocessors) (argument processing), which itself has none.

---

## Why use pyflowstep?

A fluent API makes you put every operation on the class and `return self` from each method:

```python
page.navigate("https://www.google.com").fill("input[name=q]", "pyflowstep").click("#go").wait("#result")
```

That couples *what can be done* to *one class*, you cannot store the chain, reuse half of it, build it from data, or add an operation without editing the class.

With `pyflowstep`, operations are small functions and the chain is a value:

```python
search = navigate("https://www.google.com") >> fill("input[name=q]", "pyflowstep") >> click("#go")

search(page)                      # run it
search >> wait("#result")         # extend it (a new flow, `search` is unchanged)
login >> search >> logout         # combine flows
```

…and the same flow can come from JSON:

```json
[
  {"name": "navigate", "args": ["https://www.google.com"]},
  {"name": "fill", "args": ["input[name=q]", "pyflowstep"]},
  {"name": "click", "kwargs": {"selector": "#go"}}
]
```

---

## Installation

```bash
pip install pyflowstep
```

or with uv:

```bash
uv add pyflowstep
```

Requires Python 3.12+.

---

## Quick start

```python
from typing import Any

from pyflowstep import step


class Page:
    def navigate(self, url: str) -> None:
        print(f"Navigating to {url}")

    def click(self, selector: str) -> None:
        print(f"Clicking on {selector}")

    def fill(self, selector: str, value: Any) -> None:
        print(f"Filling {selector} with {value}")


@step
def navigate(page: Page, url: str) -> Page:
    page.navigate(url)
    return page


@step
def fill(page: Page, selector: str, value: Any) -> Page:
    page.fill(selector, value)
    return page


@step
def click(page: Page, selector: str) -> Page:
    page.click(selector)
    return page


search = navigate("https://www.google.com") >> fill("input[name=q]", 1111) >> click("#go")

print(search)  # Flow(navigate >> fill >> click)
search(Page())
# Navigating to https://www.google.com
# Filling input[name=q] with 1111
# Clicking on #go
```

---

## Core concepts

### Flow

A `Flow[T]` is an immutable sequence of `T -> T` actions. Calling it threads the subject through every action in order.

```python
from pyflowstep import Flow, compose

increment = Flow[int](lambda n: n + 1)
double = Flow[int](lambda n: n * 2)

(increment >> double)(3)            # 8
(double >> increment)(3)            # 7
compose(increment, double)(3)       # 8, same as increment >> double
Flow[int]()(3)                      # 3, an empty flow is the identity
```

- `>>` accepts flows **and** plain callables, on either side: `flow >> fn`, `fn >> flow`.
- Composition flattens: `(a >> b) >> c` and `a >> (b >> c)` have the same three actions.
- Flows support `len()`, iteration, `.actions` and a readable `repr`.

### step

`@step` turns a function whose **first positional parameter is the subject** into a *step factory*. Calling the factory with the remaining arguments returns a single-step flow.

```python
from pyflowstep import step


@step
def add(total: int, amount: int) -> int:
    return total + amount


@step
def multiply(total: int, factor: int) -> int:
    return total * factor


pipeline = add(2) >> multiply(10) >> add(amount=1)
pipeline      # Flow(add >> multiply >> add)
pipeline(1)   # 31
```

`step` and `tap` do one thing: turn a function into a flow factory. Naming a step belongs to the [registry](#registry). Three markers can sit on a parameter: [`Process`](#processing-arguments) to convert the value passed for it, [`Depends`](#dependencies) to inject an object nobody passes, and [`Input`](#run-inputs) to receive a value from whoever runs the flow.

Arguments are bound against the function signature **immediately**, so mistakes fail fast:

```python
add()           # MissingArgumentError: missing a required argument: 'amount' for step 'add'
add(1, 2)       # TooManyArgumentsError: too many positional arguments for step 'add'
add(1, x=2)     # UnexpectedKeywordArgumentError: got an unexpected keyword argument 'x' ...
```

### tap

For side-effect steps on a mutable subject, `@tap` ignores the return value and passes the subject on unchanged — no more `return page` boilerplate:

```python
from pyflowstep import tap


@tap
def click(page: Page, selector: str) -> None:
    page.click(selector)
```

### override

A variant of a step keeps the step's body and parameters but changes a default: a value that used to be an argument now comes from whoever runs the flow, from a provider, or from a fixed default. Write the variant as a stub with the same parameters and decorate it with `override(original)`. The stub's body never runs; the original's does.

```python
from pyflowstep import Depends, Input, override, step


@step
def fill(form: dict, selector: str, *, value: str) -> dict:
    """Type a value into a field."""
    return {**form, selector: value}


@override(fill)
def fill_from_run(form: dict, selector: str, *, value: str = Input()) -> dict: ...


flow = fill_from_run("#date")
flow.inputs                     # frozenset({'value'})
flow({}, value="2026-11-08")    # {'#date': '2026-11-08'}
```

Any default can change. A parameter can become `Depends(...)` or get a plain default, and an `Input()` of the original can become an ordinary argument:

```python
def today() -> str:
    return "2026-11-08"


@override(fill)
def fill_today(form: dict, selector: str, *, value: str = Depends(today)) -> dict: ...


@override(fill)
def fill_blank(form: dict, selector: str, *, value: str = "") -> dict: ...


fill_today("#date")({})               # {'#date': '2026-11-08'}
fill_blank("#note")({})               # {'#note': ''}
fill_blank("#note", value="hi")({})   # {'#note': 'hi'}


@step
def greet(lines: list, name: str = Input()) -> list:
    return [*lines, f"Hello {name}"]


@override(greet)
def greet_name(lines: list, name: str) -> list: ...


greet_name("Ada")([])   # ['Hello Ada']
```

The kind of step carries over: an override of a `tap` ignores the return value too, and an override of an [`output`](#outputs) still stores its value.

```python
@tap
def log(form: dict, message: str) -> None:
    print(message, form)


@override(log)
def log_saved(form: dict, message: str = "saved") -> None: ...


log_saved()({"#a": "1"})   # prints "saved {'#a': '1'}", returns {'#a': '1'}
```

**Several variants in one flow** share an input when they keep the parameter name, because the name of an input is the name of its parameter. `override` does not rename parameters. For separate inputs, write a step whose one-line body calls the original:

```python
(fill_from_run("#from") >> fill_from_run("#to"))({}, value="2026-01-01")
# {'#from': '2026-01-01', '#to': '2026-01-01'}


@step
def fill_to(form: dict, selector: str, *, to: str = Input()) -> dict:
    return fill(selector, value=to)(form)


(fill_from_run("#from") >> fill_to("#to"))({}, value="2026-01-01", to="2026-12-31")
# {'#from': '2026-01-01', '#to': '2026-12-31'}
```

**Markers come from the stub**, so `Process` is kept only when the stub repeats it. A value that arrives through `Input` is passed as given, never processed. Dropping a `Process` without injecting the parameter is an error, because the conversion would silently disappear:

```python
@step
def repeat(lines: list, word: str, times: Annotated[int, Process(int)]) -> list:
    return [*lines, *[word] * times]


@override(repeat)
def twice(lines: list, word: str, times: Annotated[int, Process(int)] = 2) -> list: ...


@override(repeat)
def repeat_from_run(lines: list, word: str, times: int = Input()) -> list: ...


twice("hi", "3")([])                 # ['hi', 'hi', 'hi']
repeat_from_run("hi")([], times=2)   # ['hi', 'hi']  (times="2" would reach the body as a string)
```

**Docstring and name.** The stub's docstring is used, or the original's when the stub has none. The stub's name is the step name, shown in `repr` and errors. An override of an override runs the first real body:

```python
fill_blank.__doc__   # 'Type a value into a field.'


@override(fill)
def fill_date(form: dict, selector: str = "#date", *, value: str) -> dict:
    """Type the date of the voucher."""


@override(fill_date)
def fill_today_date(form: dict, selector: str = "#date", *, value: str = Depends(today)) -> dict: ...


fill_date.__doc__       # 'Type the date of the voucher.'
fill_today_date()({})   # {'#date': '2026-11-08'}
fill_from_run()         # MissingArgumentError: missing a required argument: 'selector' for step 'fill_from_run'
```

**In a registry**, `registry.override(original, name=, description=, hidden=)` registers the variant, so JSON can use it:

```python
steps = StepsRegistry[dict]()


@steps.step()
def fill(form: dict, selector: str, *, value: str) -> dict:
    """Type a value into a field."""
    return {**form, selector: value}


@steps.override(steps["fill"], name="fill_date", description="Type the voucher date.")
def fill_voucher_date(form: dict, selector: str = "#date", *, value: str = Input()) -> dict: ...


@steps.override(steps["fill"], hidden=True)
def fill_debug(form: dict, selector: str = "#debug", *, value: str = "on") -> dict: ...


flow = FlowCompiler(steps.steps).compile([{"name": "fill_date"}])
flow                           # Flow(fill_date)
flow({}, value="2026-11-08")   # {'#date': '2026-11-08'}
list(steps.steps)              # ['fill', 'fill_date']
```

In the JSON schema, `fill_date` takes only `selector` (default `"#date"`) and is described as "Type the voucher date.". `value` is a run input, so it is absent, like every input.

**Errors** are all raised when the stub is decorated:

```python
override(plain)
# InvalidStepError: 'plain' is not a step factory; override takes a factory made by step, tap, output or a StepsRegistry

@override(fill)
def fill_x(form: dict, field: str, *, value: str = Input()) -> dict: ...
# InvalidStepError: Override 'fill_x' must keep the parameters of 'fill' after the subject, with the same
# names, order and kinds: expected (selector: positional or keyword, value: keyword-only),
# got (field: positional or keyword, value: keyword-only)

@override(repeat)
def thrice(lines: list, word: str, times: int = 3) -> list: ...
# InvalidStepError: Override 'thrice' drops the Process of 'times' from 'repeat'; repeat the
# Annotated[..., Process(...)] on the stub, or inject the parameter

@override(pos)  # def pos(form: dict, selector: str, /) -> dict
def pos_input(form: dict, selector: str = Input(), /) -> dict: ...
# InvalidInputError: Step 'pos_input' cannot take positional-only parameter 'selector' as a run input

@steps.override(steps["fill"], name="fill_date")
def again(form: dict, selector: str, *, value: str = "") -> dict: ...
# StepAlreadyRegisteredError: Step 'fill_date' is already registered.
```

**Why a stub.** Python's typing cannot say "the parameters of `fill`, with `value` now optional", so the variant spells them out. In return the type checker sees the real signature, `(selector: str, *, value: str = Input()) -> Flow[dict]`, and checks every call of the variant. The parameters must match the original by name, order and kind. Only defaults and annotations may change.

---

## Registry

`StepsRegistry[T]` collects named steps for one subject type. It is the vocabulary of your flow language: the compiler and the JSON schema only know about the steps registered in it.

```python
from pyflowstep import StepsRegistry

page_steps = StepsRegistry[Page]()


@page_steps.tap()
def navigate(page: Page, url: str) -> None:
    """Open a url in the page."""
    page.navigate(url)


@page_steps.tap()
def click(page: Page, selector: str) -> None:
    """Click the element matching a CSS selector."""
    page.click(selector)


@page_steps.tap(name="type")
def fill(page: Page, selector: str, value: str) -> None:
    """Type a value into a field."""
    page.fill(selector, value)


@page_steps.tap(hidden=True)
def debug_dump(page: Page) -> None:
    """Python-only helper, never exposed to JSON."""


page_steps["click"]("#go")        # look up a step factory by name
"type" in page_steps              # True
list(page_steps.steps)            # ['navigate', 'click', 'type']  (hidden steps are excluded)

page_steps.register(lambda page: page, name="noop")  # register without a decorator
```

| Method / attribute                                               | Description                                                 |
| ---------------------------------------------------------------- | ----------------------------------------------------------- |
| `step(name=, description=, hidden=)`                             | Decorator for `(subject, ...) -> subject` functions         |
| `tap(name=, description=, hidden=)`                              | Decorator for side-effect functions (return value ignored)  |
| `register(fn, *, name=, description=, hidden=, passthrough=)`    | Register without decorator syntax                           |
| `override(original, name=, description=, hidden=)`               | Decorator registering a variant of a step, see [override](#override) |
| `chain(name=, description=, hidden=)`                            | Decorator registering a function that builds a flow, see [Chains](#chains) |
| `output(name, step_name=, description=, hidden=)`                | Decorator registering a step whose return value is a run input, see [Outputs](#outputs) |
| `steps`                                                          | Read-only mapping of visible steps                          |
| `registry[name]`, `name in registry`, `len()`, iteration         | Lookup over visible steps                                   |

- `name` is the step's public name: the compiler looks it up, and `repr` and argument errors show it (`Flow(type)` above, not `Flow(fill)`).
- `description` overrides the docstring, which becomes the step description in the JSON schema.

### Chains

A chain is a named step made of other steps. There are four ways to reuse a sequence of steps, and each one adds something to the one before:

```python
from pyflowstep import Flow, chain, step


@step
def fill(form: dict, selector: str, value: str) -> dict:
    return {**form, selector: value}


@step
def click(form: dict, selector: str) -> dict:
    return {**form, "clicked": selector}


# 1. a Flow value: fine when nothing varies
save = fill("#status", "draft") >> click("#save")


# 2. a plain function: takes arguments, but errors are plain TypeErrors
def submit(date_selector: str) -> Flow[dict]:
    return fill(date_selector, "2026-11-08") >> click("#save")


# 3. @chain: arguments are bound and processed like a step's
@chain
def submit_voucher(date_selector: str) -> Flow[dict]:
    """Fill the date and save."""
    return fill(date_selector, "2026-11-08") >> click("#save")


submit_voucher("#date")({})   # {'#date': '2026-11-08', 'clicked': '#save'}
submit()                      # TypeError: submit() missing 1 required positional argument: 'date_selector'
submit_voucher()              # MissingArgumentError: missing a required argument: 'date_selector' for step 'submit_voucher'
```

4. `@registry.chain(name=, description=, hidden=)` also registers the chain, so JSON can use it ([below](#chains-from-json)).

A chain's factory takes every parameter as a build argument. There is no subject. The flow it returns **stays flat**, so chains sit next to steps and inside other chains, and `repr` lists the underlying steps:

```python
@chain
def open_and_submit(date_selector: str) -> Flow[dict]:
    return click("#new") >> submit_voucher(date_selector)


flow = open_and_submit("#date") >> fill("#note", "done")
flow       # Flow(click >> fill >> click >> fill)
flow({})   # {'clicked': '#save', '#date': '2026-11-08', '#note': 'done'}
```

**A chain's inputs are its steps' inputs.** They show in `flow.inputs` and are checked before the first step runs:

```python
@step
def fill_number(form: dict, selector: str, number: str = Input()) -> dict:
    return {**form, selector: number}


@chain
def submit_numbered() -> Flow[dict]:
    return fill_number("#no") >> click("#save")


flow = click("#new") >> submit_numbered()
flow.inputs               # frozenset({'number'})
flow({})                  # MissingInputError: missing run input 'number' for step 'fill_number'; run the flow as flow(subject, number=...)
flow({}, number="V-7")    # {'clicked': '#save', '#no': 'V-7'}
```

**`Process` works on chain arguments**, like on step arguments:

```python
@chain
def fill_rows(selector: str, count: Annotated[int, Process(int)]) -> Flow[dict]:
    return Flow(*(action for i in range(count) for action in fill(f"{selector}{i}", "x")))


fill_rows("#row", "2")({})   # {'#row0': 'x', '#row1': 'x'}
```

#### Chains from JSON

```python
steps = StepsRegistry[dict]()
# ... fill and click registered with @steps.step()


@steps.chain(description="Fill the date and save.")
def submit(date_selector: str = "#date") -> Flow[dict]:
    return steps["fill"](date_selector, "2026-11-08") >> steps["click"]("#save")


flow = FlowCompiler(steps.steps).compile([
    {"name": "submit"},
    {"name": "submit", "kwargs": {"date_selector": "#due"}},
])
flow       # Flow(fill >> click >> fill >> click)
flow({})   # {'#date': '2026-11-08', 'clicked': '#save', '#due': '2026-11-08'}
```

In the JSON schema a chain looks like a step: `submit` takes `date_selector` in `args` or `kwargs` (a string, default `"#date"`) and is described as "Fill the date and save.".

**Errors.** Bad arguments raise the usual `ArgumentError`, with the JSON path when compiled. `Input` and `Depends` belong on the steps a chain returns, not on the chain:

```python
FlowCompiler(steps.steps).compile([{"name": "click", "args": ["#new"]}, {"name": "submit", "args": ["#a", "#b"]}])
# TooManyArgumentsError: too many positional arguments for step 'submit'
# at $[1]

@chain
def submit_input(date: str = Input()) -> Flow[dict]: ...
# InvalidStepError: Chain 'submit_input' cannot take 'date' as Input; a chain has no run of its own,
# declare it on the steps the chain returns   (the same for Depends)

@chain
def forgot(selector: str):
    fill(selector, "x")          # no return


forgot("#a")
# InvalidStepError: Chain 'forgot' must return a Flow, got None

@steps.chain(name="click")
def again() -> Flow[dict]: ...
# StepAlreadyRegisteredError: Step 'click' is already registered.
```

**A chain or a step that calls flows?** Use a chain when the sequence is fixed once its arguments are known: it stays flat, its inputs are checked up front, and JSON sees its steps in `repr` and errors. Write a step whose body runs flows, `(a() >> b())(page)`, only when the body has to decide at run time what to do, such as looping until a page shows a result. Such a step is one opaque step to the flow around it.

---

## Processing arguments

Flows often come from JSON or a web form, where every value arrives as a string, number, boolean, list, object or null, while your steps want dates, decimals, enums, clean text or loaded data. Mark the parameter with the `Process(fn)` marker of [pyargprocessors](https://github.com/aaltatan/pyargprocessors) inside `Annotated`, and `fn` is applied to the value that is passed for it:

```python
from typing import Annotated

from pyargprocessors import Process


@page_steps.tap()
def wait(page: Page, selector: str, timeout: Annotated[float, Process(float)] = 5.0) -> None:
    page.wait(selector, timeout)


wait("#result", "10")   # timeout is 10.0
```

```json
{"name": "wait", "args": ["#result", "10"]}
```

The marker sits next to the parameter it changes, and it works with the plain `@step` and `@tap` decorators too; no registry is required. `pyargprocessors` is installed together with `pyflowstep`.

### Name it once, reuse it

A `type` alias gives a processed type a name, so many steps can share it:

```python
from decimal import Decimal

type Money = Annotated[Decimal, Process(Decimal)]
type Text = Annotated[str, Process(str.strip), Process(str.lower)]   # several processors run left to right


@catalog_steps.step()
def price_between(products: Catalog, low: Money, high: Money) -> Catalog: ...


@catalog_steps.step()
def search(products: Catalog, text: Text, min_stars: Annotated[float, Process(float)] = 0) -> Catalog: ...
```

### Loading data from a value in the JSON

A processor can be any function of one value, so the JSON can send a path and the step can receive what was loaded from it:

```python
def load_names(path: str) -> frozenset[str]:
    return frozenset(Path(path).read_text().splitlines())


type Names = Annotated[frozenset[str], Process(load_names)]


@catalog_steps.step()
def exclude(products: Catalog, names: Names) -> Catalog:
    return tuple(product for product in products if product.name not in names)
```

```json
{"name": "exclude", "args": ["discontinued.txt"]}
```

The file is read **once, when the flow is built**. Every run of that flow reuses the loaded names. For an object that must be fresh on every run, or that the JSON must not choose at all, use a [dependency](#dependencies) instead.

### `*args` and `**kwargs`

On `*args` the processor applies to each item, and on `**kwargs` to each value:

```python
@barista.step()
def top_with(drink: Drink, *toppings: Text) -> Drink: ...


@catalog_steps.step()
def where(products: Catalog, *, in_stock: Annotated[bool, Process(to_bool)] = False, **fields: Text) -> Catalog: ...


where(in_stock="yes", brand=" SONIC ")   # in_stock=True, brand="sonic"
```

### Lists and nulls

A JSON list or a `null` is one value, so `Process(int)` would be applied to the whole of it. The `each` and `optional` helpers of `pyargprocessors` build the processor you need:

```python
from pyargprocessors import each, optional

type Prices = Annotated[list[Decimal], Process(each(Decimal))]   # every item of a list
type Limit = Annotated[int | None, Process(optional(int))]       # null is kept as None
```

### Which values are processed, and when

- Only values that are **actually passed**, positionally or by keyword. Default values are never processed.
- Processing runs **once per factory call**, while the flow is being built, not every time the flow runs.
- Arguments are checked against the step's signature first, so a missing or extra argument is reported as an `ArgumentError` before any processor runs.

```python
flow = add_syrup("  Vanilla ", "2")   # processed here: flavor="vanilla", pumps=2
flow(drink)                           # the step runs with the processed values
flow(another_drink)                   # nothing is processed again
```

### Errors

A processor that fails on a value raises `ProcessArgumentError`, chained to the original exception. It has the name of the argument (`error.argument`), the value as it was passed (`error.value`) and the original exception (`error.cause`). Inside a compiled flow it also carries the JSON path of the step, so a bad value is found before anything runs:

```text
pyargprocessors.exceptions.ProcessArgumentError: Argument 'limit' with value 'two' failed to process, invalid literal for int() with base 10: 'two'
at $[1]
```

A marker that cannot work raises `InvalidProcessorError` when the step is created:

| Problem                                              | Example                                              |
| ---------------------------------------------------- | ---------------------------------------------------- |
| The processor is not callable                        | `Process("int")`                                     |
| The marker is used as a default value                | `amount: int = Process(int)`, write `Annotated[int, Process(int)]` |
| The marker is on the subject (the first parameter)   | `def step(page: Annotated[Page, Process(...)], ...)` |
| One parameter has both `Process` and `Depends` or `Input` | nothing is passed for it, so there is nothing to process |

Both exceptions come from `pyargprocessors`. They are `TypeError`s and derive from `PyargprocessorsError`, **not** from `PyflowstepError`:

```python
from pyargprocessors import ProcessArgumentError

try:
    flow = compiler.compile(definition)
except ProcessArgumentError as error:    # a value in the definition is wrong
    ...
except PyflowstepError as error:         # the definition itself is wrong
    ...
```

### In the JSON schema

A processed parameter is described by **what the JSON must send**. The schema takes it from the type hint of the processor's own parameter (`load_names(path: str)` means `string`). If the processor has no type hint, as with `int`, `Decimal` or an enum class, the schema falls back to the parameter's type.

| Parameter                                    | Schema                          |
| -------------------------------------------- | ------------------------------- |
| `names: Annotated[frozenset[str], Process(load_names)]` | `{"type": "string"}`   |
| `limit: Annotated[int, Process(int)]`        | `{"type": "integer"}`           |
| `kind: Annotated[Milk, Process(Milk)]`       | `{"enum": ["whole", "oat", "almond"], "type": "string"}` |

### Upgrading

Argument processing moved to `pyargprocessors`, and `Parse` became its `Process`:

| Before                                            | Now                                                     |
| ------------------------------------------------- | ------------------------------------------------------- |
| `from pyflowstep import Parse`                    | `from pyargprocessors import Process`                   |
| `Annotated[float, Parse(float)]`                  | `Annotated[float, Process(float)]`                      |
| `pyflowstep.ParseArgumentError`                   | `pyargprocessors.ProcessArgumentError`, no longer a `PyflowstepError` |
| `pyflowstep.InvalidParserError`                   | `pyargprocessors.InvalidProcessorError`, no longer a `PyflowstepError` |
| `... failed to parse, ...` in the message         | `... failed to process, ...`                            |

Coming from 0.1, the `processors=` option of the registry is gone as well. Move each entry onto its parameter:

| 0.1                                               | Now                                                     |
| ------------------------------------------------- | ------------------------------------------------------- |
| `processors={"timeout": float}`                   | `timeout: Annotated[float, Process(float)]`             |
| `processors=normalize` (every argument)           | annotate each parameter, usually with a shared alias such as `Text` |
| `processors={"pumps": int, ...: normalize}`       | `pumps: Annotated[int, Process(int)]`, the others `Text` |
| `...` covering `**kwargs`                         | `**fields: Text`                                        |
| `InvalidProcessorsError`                          | removed; a key can no longer be misspelled              |

See [`examples/processing.py`](examples/processing.py) for every form working together on data from a web form.

---

## Dependencies

Some steps need objects that cannot be written in JSON: a mailer, a database session, an API client. Mark the parameter with `Depends(provider)` and it stops being an argument. Nobody passes it, neither Python nor JSON; `provider` is called to produce it when the flow runs.

```python
from pyflowstep import Depends, StepsRegistry

steps = StepsRegistry[Order]()


def get_mailer() -> Mailer:
    return Mailer("smtp.example.com")


@steps.tap()
def send_email(order: Order, template: str, mailer: Mailer = Depends(get_mailer)) -> None:
    mailer.send(order.customer, template)


send_email("receipt")   # only `template` is an argument
```

```json
[{"name": "send_email", "args": ["receipt"]}]
```

It works the same with the plain `@step` and `@tap` decorators; no registry is required.

### Always the default value

`Depends(...)` is written as the default value of the parameter, in steps and in providers. To name a dependency once and reuse it in many steps, keep the marker in a constant:

```python
# inline: quick, for a one-off
def send_email(order: Order, template: str, mailer: Mailer = Depends(get_mailer)) -> None: ...


# a constant: name the dependency once, reuse it
MAILER = Depends(get_mailer)

def send_email(order: Order, template: str, mailer: Mailer = MAILER) -> None: ...
def send_invoice(order: Order, mailer: Mailer = MAILER) -> None: ...
```

It is never written inside `Annotated`. Only a default value tells a type checker that the parameter is not passed, so `send_email("receipt")` type-checks; `Annotated[Mailer, Depends(get_mailer)]` raises `InvalidDependencyError` when the step is created.

### One object per flow run

A flow run is one scope, like one request in a web framework. A provider is called **at most once per run**, and every step of that run receives the same object. The next run starts fresh.

A provider written as a generator is cleaned up when the run ends. If a step fails, the error is raised inside the provider at its `yield`, so it can roll back:

```python
from collections.abc import Iterator


def get_session() -> Iterator[Session]:
    session = Session()
    try:
        yield session          # shared by every step of this run
    except Exception:
        session.rollback()     # a step failed
        raise
    else:
        session.commit()       # the whole flow succeeded
    finally:
        session.close()


SESSION = Depends(get_session)


@steps.tap()
def save(order: Order, session: Session = SESSION) -> None:
    session.add(order)


@steps.tap()
def audit(order: Order, action: str, session: Session = SESSION) -> None:
    session.add(AuditRow(order.id, action))   # the same session `save` used


flow = save() >> audit("fulfilled")

flow(order_1)   # session A: opened, used twice, committed, closed
flow(order_2)   # session B
```

Cleanups run in reverse order, and a provider cannot swallow a step's error: it is always re-raised. Flows nested inside a flow, or called from inside a step, join the run that is already open.

### Sub-dependencies

A provider's own parameters can use `Depends` too:

```python
def get_settings() -> Settings:
    return Settings()


def get_mailer(settings: Settings = Depends(get_settings)) -> Mailer:
    return Mailer(settings.smtp_host)
```

Any other provider parameter must have a default. A provider can be any callable: a function, a class, a `functools.partial`, or an object with `__call__`.

### Replacing a dependency in tests

```python
from pyflowstep import override_dependencies

with override_dependencies({get_mailer: FakeMailer}):
    flow(order)          # every step that asked for get_mailer gets a FakeMailer

flow(order)              # the real mailer again
```

Keys are the original providers and values the providers to call instead. Nested blocks add up, and everything is restored when the block exits, even after an error.

### Rules

- **Invisible to JSON.** A dependency is absent from the JSON schema, cannot be processed, and passing one raises `UnexpectedKeywordArgumentError` (or `TooManyArgumentsError`) when the flow is built.
- **Declarations are checked early**, when the step is created, with `InvalidDependencyError`: a provider that is not callable, is circular, or has a required parameter that is not a dependency; a dependency on the subject or on a positional-only parameter; `Depends` written inside `Annotated`.
- **Providers run late**, when the flow runs. An error inside a provider surfaces then, not at compile time.
- A dependency is for an object the JSON must **not** choose. When the JSON should pick one by name (`"via": "email"`), use [`Process`](#processing-arguments) with a function that turns the name into the object.

Using ruff? Its `B008` rule flags function calls in argument defaults. Tell it `Depends` is a marker:

```toml
[tool.ruff.lint.flake8-bugbear]
extend-immutable-calls = ["pyflowstep.Depends"]
```

See [`examples/dependencies.py`](examples/dependencies.py) for a complete flow with a mailer, a per-run session and a test override.

### Upgrading from 0.4

`Annotated[T, Depends(provider)]` is gone, for the same reason `Input[T]` went in 0.4: a type checker reported `send_email("receipt")` as missing its `mailer` argument. Move the marker to the default value, in steps and in providers:

| 0.4                                                       | 0.5                                          |
| --------------------------------------------------------- | -------------------------------------------- |
| `mailer: Annotated[Mailer, Depends(get_mailer)]`          | `mailer: Mailer = Depends(get_mailer)`       |
| `type MailerDep = Annotated[Mailer, Depends(get_mailer)]` | `MAILER = Depends(get_mailer)`               |
| `mailer: MailerDep`                                       | `mailer: Mailer = MAILER`                    |

---

## Run inputs

Some values exist only where the flow is run: the logged-in user, the record being processed, credentials read from a prompt. No provider can build them and JSON must not hold them. Give the parameter the default `Input()` and the caller supplies it, by name, when it runs the flow:

```python
from pyflowstep import Input, StepsRegistry

steps = StepsRegistry[Page]()


@steps.tap()
def login(page: Page, url: str, credentials: Credentials = Input()) -> None:
    page.navigate(url)
    page.fill("#user", credentials.username)
    page.fill("#password", credentials.password)


@steps.tap()
def fill_year(page: Page, selector: str, voucher: Voucher = Input()) -> None:
    page.fill(selector, voucher.year)


flow = login("https://example.com/login") >> fill_year("#year")   # inputs are not arguments

flow(page, credentials=credentials, voucher=voucher)              # they are given to the run
```

```json
[
  {"name": "login", "args": ["https://example.com/login"]},
  {"name": "fill_year", "args": ["#year"]}
]
```

The name of the input is the name of the parameter, and every step that declares it receives the same value.

`Input()` is written as the default value, never inside `Annotated`, so that type checkers see the parameter as optional and accept `fill_year("#year")`.

### Checked before anything runs

A flow knows the inputs its steps require, and checks them before the first step runs. A forgotten input never fails halfway through:

```python
flow.inputs   # frozenset({'credentials', 'voucher'})

flow(page, credentials=credentials)
# MissingInputError: missing run input 'voucher' for step 'fill_year'; run the flow as flow(subject, voucher=...)
```

### Typed calls

A flow is typed as `flow(obj, **inputs: Any)`, because Python's typing cannot merge the inputs of several steps into one signature. To have a type checker check the inputs, write the signature once as a stub and decorate it with `flow.signature`:

```python
@flow.signature
def log_in_and_fill(page: Page, /, *, credentials: Credentials, voucher: Voucher) -> Page: ...

log_in_and_fill(page, credentials=credentials)   # type checker: missing argument "voucher"
```

The subject is positional-only and every input keyword-only. Required inputs have no default; optional ones need one (`= ...` is fine), but the step's own default is what gets used. The stub is checked against the steps when it is decorated, and `InvalidInputError` names any input that is missing, unknown, or required when it should be optional (or the reverse). Types are not compared, only names. mypy reports a `...` body as `empty-body`; write `raise NotImplementedError` instead if that bothers you.

### Rules

- **Invisible to JSON**, like a dependency: an input is absent from the JSON schema, cannot be processed, and passing one while building the flow raises `UnexpectedKeywordArgumentError` (or `TooManyArgumentsError`).
- **`Input(default=...)` makes the input optional**: with `note: str = Input(default="")`, `""` is used when the caller passes no `note`. Optional inputs are not listed in `flow.inputs`.
- **Extra inputs are ignored**, so one caller can run different flows, each using the inputs it needs.
- **Nested flows see the inputs of the run they join.** A flow called from inside a step can be given inputs of its own, `inner(page, voucher=other)`; they are laid over the outer ones for that call only.
- **Declarations are checked early**, when the step is created, with `InvalidInputError`: an input on the subject, on a positional-only parameter, or written inside `Annotated`. A provider cannot take an input.

Using ruff? Add `Input` next to `Depends` for its `B008` rule:

```toml
[tool.ruff.lint.flake8-bugbear]
extend-immutable-calls = ["pyflowstep.Depends", "pyflowstep.Input"]
```

### Upgrading from 0.3

`Input[T]` is gone, because a type checker reported `fill_year("#year")` as missing its `voucher` argument. Move the marker to the default value:

| 0.3                                | 0.4                                      |
| ---------------------------------- | ---------------------------------------- |
| `voucher: Input[Voucher]`          | `voucher: Voucher = Input()`             |
| `note: Input[str] = ""`            | `note: str = Input(default="")`          |

### Outputs

Sometimes a later step needs a value that an earlier step finds: a number read from the page, an id returned by a save. Decorate the earlier step with `output(name)`. Its return value becomes the run input `name`, and the subject passes through unchanged, like a `tap`. The later step reads it with an ordinary `Input()`, so it stays reusable with or without a producer:

```python
from pyflowstep import Input, output, step


@output("number")
def read_number(form: dict) -> str:
    return form["#no"]


@step
def fill(form: dict, selector: str, number: str = Input()) -> dict:
    return {**form, selector: number}


flow = read_number() >> fill("#target")
flow({"#no": "V-7"})   # {'#no': 'V-7', '#target': 'V-7'}
flow.inputs            # frozenset(): the caller no longer gives `number`
```

One output can be read by several steps, a flow can have several outputs, and an optional `Input(default=...)` takes the produced value once there is one:

```python
@step
def note(form: dict, number: str = Input()) -> dict:
    return {**form, "note": f"copied {number}"}


@output("year")
def read_year(form: dict) -> str:
    return form["#date"][:4]


@step
def fill_ref(form: dict, number: str = Input(), year: str = Input()) -> dict:
    return {**form, "#ref": f"{year}/{number}"}


@output("user")
def read_user(form: dict) -> str:
    return form["#owner"]


@step
def stamp(form: dict, user: str = Input(default="system")) -> dict:
    return {**form, "#by": user}


(read_number() >> fill("#target") >> note())({"#no": "V-7"})
# {'#no': 'V-7', '#target': 'V-7', 'note': 'copied V-7'}
(read_number() >> read_year() >> fill_ref())({"#no": "V-7", "#date": "2026-11-08"})
# {'#no': 'V-7', '#date': '2026-11-08', '#ref': '2026/V-7'}
(stamp() >> read_user() >> stamp())({"#owner": "ada"})
# {'#owner': 'ada', '#by': 'ada'}   (the first stamp wrote 'system', the second 'ada')
```

A producer works anywhere in the flow, also inside a [chain](#chains) or from JSON. `registry.output(name, step_name=, description=, hidden=)` registers one. Its first argument is the output name, so the step name is the keyword `step_name`:

```python
@chain
def copy_number(selector: str) -> Flow[dict]:
    return read_number() >> fill(selector)


(copy_number("#target") >> note())({"#no": "V-7"})
# {'#no': 'V-7', '#target': 'V-7', 'note': 'copied V-7'}

steps = StepsRegistry[dict]()


@steps.output("number", step_name="read_number")
def read_no(form: dict) -> str:
    """Read the voucher number."""
    return form["#no"]


# ... fill registered with @steps.step()

flow = FlowCompiler(steps.steps).compile([
    {"name": "read_number"},
    {"name": "fill", "args": ["#target"]},
])
flow({"#no": "V-7"})   # {'#no': 'V-7', '#target': 'V-7'}
```

**Order is checked before anything runs.** A step that reads `number` before the producer still needs it from the caller, so a wrong order fails before the first step:

```python
flow = save() >> fill("#target") >> read_number()
flow.inputs            # frozenset({'number'})
flow({"#no": "V-7"})   # MissingInputError: missing run input 'number' for step 'fill'; run the flow as flow(subject, number=...)
                       # and `save` never ran
```

`Flow.signature` follows the same rule, so a produced name must not be declared:

```python
@(read_number() >> fill("#target")).signature
def copy(form: dict, /) -> dict: ...


@(read_number() >> fill("#target")).signature
def copy2(form: dict, /, *, number: str) -> dict: ...
# InvalidInputError: Signature 'copy2' does not match the flow: unknown input 'number'
```

**Replacing values.** A later output of the same name replaces the earlier one, and a produced value replaces one the caller passed, both only for the steps after the producer. A step that needs its own output name reads it before it writes it, so the caller must give it:

```python
(read_number() >> fill("#a") >> set_no("V-8") >> read_number() >> fill("#b"))({"#no": "V-7"})
# {'#no': 'V-8', '#a': 'V-7', '#b': 'V-8'}

(fill("#a") >> read_number() >> fill("#b"))({"#no": "V-7"}, number="given")
# {'#no': 'V-7', '#a': 'given', '#b': 'V-7'}


@output("number")
def next_number(form: dict, number: str = Input()) -> str:
    return str(int(number) + 1)


(next_number() >> fill("#no")).inputs   # frozenset({'number'})
(next_number() >> fill("#no"))({}, number="41")   # {'#no': '42'}
```

**Scope.** An output lives in its run. It is gone when the run ends, so the next run starts without it. A nested flow called with inputs of its own gets an overlay, so what it outputs ends with that call:

```python
inner = read_number() >> fill("#inner")


@step
def run_inner(form: dict) -> dict:
    return inner({**form, "#no": "INNER"}, mode="nested")


(read_number() >> run_inner() >> fill("#outer"))({"#no": "V-7"})
# {'#no': 'INNER', '#inner': 'INNER', '#outer': 'V-7'}
```

**An output step is a step**, so it can take its own `Input` and `Depends`. A producer that returns `None` produces `None`:

```python
@output("number")
def make_number(form: dict, year: str = Input(), prefix: str = Depends(get_prefix)) -> str:
    return f"{prefix}-{year}-{len(form)}"


(make_number() >> fill("#no"))({}, year="2026")   # {'#no': 'INV-2026-0'}


@output("number")
def maybe_number(form: dict) -> str | None:
    return form.get("#no")


@step
def fill_or_skip(form: dict, number: str | None = Input()) -> dict:
    return form if number is None else {**form, "#target": number}


(maybe_number() >> fill_or_skip())({})   # {}
```

The output name must be usable as a parameter name, and it is checked when the step is decorated:

```python
output("voucher-number")
# InvalidInputError: Invalid output name: 'voucher-number', it names the run input that later steps read,
# so it must be a valid parameter name: an identifier that is not a keyword
output("class")   # the same error: a keyword
```

### `Input`, `Depends` or an output?

| The object…                                                  | Use                                   |
| ------------------------------------------------------------ | ------------------------------------- |
| is written in the flow definition (a selector, a limit)      | a plain argument, with `Process` if needed |
| can be built by a function, the same way for every caller (a mailer, a session) | `Depends(provider)`     |
| is only known to whoever runs the flow (a user, a record, credentials) | `Input()`                   |
| is found by an earlier step of the same run (a number read from the page, a saved id) | `output(name)` on that step, `Input()` on the readers |

---

## Compiling flows from JSON

`FlowCompiler` turns a flow definition — a list of step dictionaries — into a `Flow`.

```python
from pyflowstep import FlowCompiler

compiler = FlowCompiler(page_steps.steps)

login = compiler.compile(
    [
        {"name": "navigate", "args": ["https://example.com/login"]},
        {"name": "type", "args": ["#email", "ada@example.com"]},
        {"name": "click", "kwargs": {"selector": "button[type=submit]"}},
    ],
)

login(Page())
```

`compile` takes already-parsed data, so a database row or an API payload works too.

### Flow files

`compile_file` picks a parser by the file suffix. JSON and TOML are built in:

```python
login = compiler.compile_file("login.json")  # a list of steps, as above
login = compiler.compile_file("login.toml")
```

A TOML flow lists its steps as `[[steps]]` tables:

```toml
[[steps]]
name = "navigate"
args = ["https://example.com/login"]

[[steps]]
name = "click"
kwargs = { selector = "button[type=submit]" }
```

Add any other format by registering a parser that turns the file text into a list of steps:

```python
import yaml

from pyflowstep import register_flow_format

register_flow_format(".yaml", yaml.safe_load)
register_flow_format(".yml", yaml.safe_load)
```

An unregistered suffix raises `UnsupportedFlowFormatError`; a file that does not parse raises `InvalidFlowDefinitionError`. Errors from a file carry an extra `in <path>` note. `load_flow_definition(path)` returns the parsed data without compiling it.

Each step dictionary has this shape — only `name` is required:

```json
{"name": "fill", "args": ["#email"], "kwargs": {"value": "ada@example.com"}}
```

Everything is validated while compiling, before any step runs. Errors carry a note with their JSON path:

```text
pyargprocessors.exceptions.ProcessArgumentError: Argument 'url' with value 'http://insecure.example.com' failed to process, only https urls are allowed
at $[1]
```

| Problem                                          | Exception                                              |
| ------------------------------------------------ | ------------------------------------------------------ |
| Not a list / malformed step dict / bad file      | `InvalidFlowDefinitionError`                           |
| Unknown or hidden step name                      | `StepDoesNotExistError` (lists the available steps)    |
| Missing / extra / duplicated arguments           | `MissingArgumentError`, `TooManyArgumentsError`, ...   |
| A processor rejects a value                      | `ProcessArgumentError` (from `pyargprocessors`)        |

A compiled flow is an ordinary `Flow`, so it composes with Python steps: `login >> click("#profile")`.

---

## JSON Schema

Describe your flow language so a UI, a validator, or an LLM can produce valid flows:

```python
import json

from pyflowstep import get_flow_json_schema, get_step_json_schema

print(json.dumps(get_flow_json_schema(page_steps.steps), indent=2))
```

Output, trimmed to the `click` step:

```json
{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "array",
  "items": {
    "oneOf": [
      {
        "type": "object",
        "properties": {
          "name": {"const": "click"},
          "args": {"type": "array", "prefixItems": [{"type": "string"}], "items": false},
          "kwargs": {
            "type": "object",
            "properties": {"selector": {"type": "string"}},
            "required": [],
            "additionalProperties": false
          }
        },
        "required": ["name"],
        "additionalProperties": false,
        "description": "Click the element matching a CSS selector."
      }
    ]
  }
}
```

Supported annotations: `str`, `int`, `float`, `bool`, `None`, `Decimal`, `datetime`, `date`, `time`, `UUID`, `list`/`tuple`/`set`/`frozenset`, `dict`, `Literal`, `Enum`, unions, `Annotated`, `TypedDict`, and `type` aliases. Anything else maps to `{}` (any value). JSON-compatible default values are included as `default`.

---

## Examples

Runnable examples live in [`examples/`](examples):

- [`examples/browser.py`](examples/browser.py) — the `Page` automation above: `tap` steps, an https-only processor, reusable sub-flows and a JSON login scenario.

  ```bash
  uv run python -m examples.browser
  ```

- [`examples/coffee.py`](examples/coffee.py) — a coffee shop where every step returns a **new** immutable `Drink`. The menu is JSON, orders are customized with Python steps, and a hidden staff-only `discount` step is invisible to the menu.

  ```bash
  uv run python -m examples.coffee
  ```

  ```python
  order("latte", size("L"), add_syrup("vanilla")).describe()
  # 'L espresso, 1 shot(s), 200ml whole milk, 1x vanilla syrup -> $3.65'
  ```

- [`examples/processing.py`](examples/processing.py) — every way to use `Process`, side by side. Shop filters arrive from a web form as raw strings (`"4"`, `"yes"`, `" SONIC "`, a file path) and are turned into typed, clean arguments, including a set of names loaded from that path:

  ```bash
  uv run python -m examples.processing
  ```

  ```python
  type Text = Annotated[str, Process(str.strip), Process(str.lower)]
  type Names = Annotated[frozenset[str], Process(load_names)]

  @catalog_steps.step()
  def where(products: Catalog, *, in_stock: Annotated[bool, Process(to_bool)] = False, **fields: Text) -> Catalog: ...

  @catalog_steps.step()
  def exclude(products: Catalog, names: Names) -> Catalog: ...
  ```

- [`examples/dependencies.py`](examples/dependencies.py) — steps that need objects JSON cannot describe. A fulfilment flow stored as JSON uses a mailer (with a sub-dependency) and a database session that is opened once per run, shared by two steps, then committed or rolled back. Also shows swapping the mailer for a fake in a test:

  ```bash
  uv run python -m examples.dependencies
  ```

  ```python
  @fulfilment_steps.tap()
  def send_email(order: Order, template: str, mailer: Mailer = Depends(get_mailer)) -> None: ...

  @fulfilment_steps.tap()
  def save(order: Order, session: Session = SESSION) -> None: ...
  ```

### Working with pyspecification and pyformula

Predicates and formulas are just callables, so they plug straight into steps:

```python
from dataclasses import dataclass, replace
from decimal import Decimal

from pyformula import variable
from pyspecification import object_rule
from pyflowstep import step


@dataclass(frozen=True)
class Invoice:
    subtotal: Decimal
    tax: Decimal = Decimal(0)
    notes: tuple[str, ...] = ()


@object_rule()
def is_large(invoice: Invoice, limit: Decimal) -> bool:
    return invoice.subtotal >= limit


@variable()
def subtotal(invoice: Invoice) -> Decimal:
    return invoice.subtotal


@step
def note_if(invoice: Invoice, rule, note: str) -> Invoice:
    return replace(invoice, notes=(*invoice.notes, note)) if rule(invoice) else invoice


@step
def apply_tax(invoice: Invoice, formula) -> Invoice:
    return replace(invoice, tax=Decimal(str(formula(invoice))))


checkout = note_if(is_large(Decimal(100)), "needs approval") >> apply_tax(round(subtotal * 0.15, 2))
```

---

## API reference

| Name                                                         | Kind      | Description                                                     |
| ------------------------------------------------------------ | --------- | --------------------------------------------------------------- |
| `Flow[T]`                                                    | class     | Immutable, callable sequence of actions; composes with `>>`     |
| `compose(*actions)`                                          | function  | Combine actions and flows into one flat flow                    |
| `step` / `tap`                                               | decorator | Turn a function into a step factory                             |
| `override(original)`                                         | decorator | A variant of a step with other defaults, see [override](#override) |
| `chain`                                                      | decorator | A named step factory made of other steps, see [Chains](#chains) |
| `output(name)`                                               | decorator | A step whose return value becomes the run input `name`, see [Outputs](#outputs) |
| `StepsRegistry[T]`                                           | class     | Named collection of steps                                       |
| `pyargprocessors.Process(fn)`                                | marker    | Apply `fn` to the value passed for a parameter, see [Processing arguments](#processing-arguments) |
| `Depends(provider)`                                          | marker    | Inject a parameter by calling `provider`, see [Dependencies](#dependencies) |
| `override_dependencies(mapping)`                             | context manager | Replace providers inside a `with` block, for tests        |
| `Input(default=...)`                                         | marker    | Receive a parameter from the caller of the flow, `flow(subject, name=value)`, see [Run inputs](#run-inputs) |
| `Flow.inputs`                                                | property  | The names of the inputs a flow requires                         |
| `Flow.signature(stub)`                                       | decorator | Give a flow a typed call signature, checked against its inputs, see [Typed calls](#typed-calls) |
| `FlowCompiler[T](steps)`                                     | class     | `compile(definition)` turns parsed step dicts into a flow       |
| `validate_step_dict(item, path)`                             | function  | Validate one step dictionary                                    |
| `get_json_schema(annotation)`                                | function  | JSON Schema of a type annotation                                |
| `get_step_json_schema(name, step)`                           | function  | JSON Schema of one step dictionary                              |
| `get_flow_json_schema(steps)`                                | function  | JSON Schema of a whole flow definition                          |

All exceptions derive from `PyflowstepError`; argument errors, `InvalidDependencyError`, `InvalidInputError` and `MissingInputError` also derive from `TypeError`, definition errors from `ValueError`, and `StepDoesNotExistError` from `LookupError`. The exceptions of argument processing, `ProcessArgumentError` and `InvalidProcessorError`, belong to `pyargprocessors`: they are `TypeError`s too, but derive from `PyargprocessorsError`.

---

## Development

```bash
uv sync
uv run pytest --cov --cov-report term-missing
```

The test suite also runs every docstring example (`--doctest-modules`).

## License

GPL-3.0
