Metadata-Version: 2.4
Name: vcti-flow
Version: 1.0.0
Summary: Payload-agnostic, generically typed framework for composing flow graphs from sources, transformers, reducers, sinks, and combinators.
Author: Visual Collaboration Technologies Inc.
Requires-Python: <3.15,>=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: test
Requires-Dist: pytest; extra == "test"
Requires-Dist: pytest-cov; extra == "test"
Provides-Extra: lint
Requires-Dist: ruff; extra == "lint"
Provides-Extra: typecheck
Requires-Dist: mypy; extra == "typecheck"
Dynamic: license-file

# Flow

Payload-agnostic, generically typed framework for composing flow graphs from sources, transformers, reducers, sinks, and combinators.

## Overview

`vcti.flow` models a computation as a right-to-left graph of nodes: a **Source**
produces a value, a **Transformer** maps one value to another, a **Reducer**
combines many values into one, and a **Sink** consumes a value and passes it
through. Calling `execute()` on the final node pulls the whole graph.

The framework is **generic over the payload type** — `Node[OutT]`, with
`Transformer[InT, OutT]` and `Reducer[InT, OutT]` carrying type transitions — so
the same graph machinery works for any payload (NumPy arrays, dataclasses,
domain objects). It never inspects payloads: the type parameters are
compile-time documentation, and there are no runtime dependencies. Fan-out and
other graph constructions live in a small **combinator** layer
(`for_each`, …) rather than as node types.

## Installation

```bash
pip install vcti-flow
```

### In `requirements.txt`

```
vcti-flow>=1.0.0
```

### In `pyproject.toml` dependencies

```toml
dependencies = [
    "vcti-flow>=1.0.0",
]
```

---

## Quick Start

```python
from vcti.flow import Source, Transformer, Reducer, Sink, for_each

# A source produces a value (0 -> 1)
class Constant(Source[int]):
    def __init__(self, value: int) -> None:
        super().__init__()
        self._value = value

    def load(self) -> int:
        return self._value

# A transformer maps one value to another (1 -> 1; type may change)
class Double(Transformer[int, int]):
    def transform(self, record: int) -> int:
        return record * 2

# A reducer combines many values into one (N -> 1)
class Sum(Reducer[int, int]):
    def reduce(self, records: list[int]) -> int:
        return sum(records)

# Compose right-to-left and execute
result = Double().connect(Constant(21)).execute()                          # 42

# Combine two sources
total = Sum().connect(Constant(1)).connect(Constant(2)).execute()          # 3

# Fan out into one flow per item (a combinator, not a node)
for key, node in for_each(Constant(3), range, Constant):
    print(key, node.execute())                                             # 0 0 / 1 1 / 2 2
```

---

## Node taxonomy

| Kind | Arity | Implement |
|------|-------|-----------|
| `Source[OutT]` | 0 → 1 | `load()` |
| `Transformer[InT, OutT]` | 1 → 1 | `transform(record)` |
| `Reducer[InT, OutT]` | N → 1 | `reduce(records)` |
| `Sink[T]` | 1 → 1 (passthrough) | `save(record)` (optional) |
| `for_each(...)` | 1 → N | combinator (graph construction) |

`Observer[OutT]` observes connect / before-execute / after-execute events
(subclass `ObserverBase` to override only the events you need);
`execute_cached()` memoises shared nodes in diamond graphs; `validate()` checks
for cycles and wrong input counts before running. Framework errors derive from
`FlowError` (`FlowWiringError`, `FlowStateError`, `FlowValidationError`).

See the docs for more: [patterns](docs/patterns.md) (practical recipes),
[design](docs/design.md) (concepts and rationale),
[extending](docs/extending.md) (adding combinators and wrappers), and
runnable [examples/](examples/).

---

## Dependencies

None. Standard library only.
