Skip to content

Pipeline Scaffold

Summary

Data-driven processing is expressed as a composition graph. The graph is built by a DI library; execution is driven by a DAG engine. Stages are pure and individually testable.

When to use a pipeline

Use a pipeline when a resource is produced by a multi-step transformation of an input (a document, a file, a message) rather than by a simple CRUD write.

Structure of a family

A family is one input shape (e.g. a file format or a message kind). Each family has the same skeleton:

1
2
3
4
5
6
7
8
pipelines/<family>/
    abc.py            # port contracts (stage interfaces + root interface)
    impl.py           # concrete stages + root implementation
    pipeline.yaml     # graph spec: nodes, edges, roots, schema
    resources/
        pipeline.yaml # runtime spec consumed by the loader
        patterns/*.yaml  # rule fragments (pattern registry)
    banks/<variant>.py   # optional variants that re-pin a few slots

Shared scaffolding

1
2
3
4
pipelines/shared/
    ports.py     # RegistryPort, FinalizePort, etc.
    loader.py    # build(root_cls) -> wires the graph + engine
    registry.py  # resolve_root(...) -> root class for a (kind, variant)

The loader

1
2
3
4
5
6
7
def build(root_cls, *, instances=None, config=None):
    # merge resources/patterns/*.yaml into a pattern registry
    # bind registry/config into each graph node by id
    # build the schema (state) class and the engine
    root = root_cls()
    root._engine = Engine(graph, state_cls, payload_cls)
    return root

The registry

A single lookup replaces hand-rolled if/else selection:

1
2
3
4
5
6
7
_FAMILIES = {
    "csv":  {"acme": {"heuristic": AcmeCsvPipeline}},
    "json": {"globex": {"heuristic": GlobexJsonPipeline}},
}

def resolve_root(kind, variant, mode="heuristic"):
    return _FAMILIES[kind][variant][mode]

Adding a new variant means adding one class and one registry entry.

Driving a pipeline

Stages are driven explicitly so intermediate state can be persisted:

1
2
3
4
pipeline = build(resolve_root(kind, variant))
content = pipeline.ingest.materialize(raw_bytes)
records = pipeline.reader.read(content)
items   = pipeline.normalize.build(records)

Testing

Because stages are pure, test each stage in isolation and test the assembled root against fixture inputs whose expected outputs are stored as golden files. A parity test replays fixtures headlessly.