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

pip install vcti-flow

In requirements.txt

vcti-flow>=2.0.0

In pyproject.toml dependencies

dependencies = [
    "vcti-flow>=2.0.0",
]

Quick Start

from vcti.flow.core 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

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 (practical recipes), design (concepts and rationale), extending (adding combinators and wrappers), and runnable examples/.

Dependencies

None. Standard library only.