Static type-checking for cyclic pipelines in Python using operator-overloaded composition

Learn how to type-check cyclic pipelines in Python with static typing: enforce adjacency via operator overloading and ensure ring closure with a processor.

Type-checking a cyclic pipeline of components in Python is tricky when you want static guarantees that adjacent elements line up and the chain wraps around cleanly. The core requirement is simple to say but hard to encode in types: the output of each element must match the input of the next one, and the last one must point back to the first.

Problem statement

Suppose we have a generic building block with input and output types, and a processor that accepts a tuple of such blocks. The expectation is that every neighbor in the tuple composes, and the chain forms a loop.

class Node[XIn, XOut]: ...

class CycleManager[...]:
    def __init__(self, items: tuple[...]): ...

For a valid setup, outputs and inputs match across neighbors and wrap:

u1 = Node[int, str](...)
u2 = Node[str, complex](...)
u3 = Node[complex, int](...)

mgr = CycleManager((u1, u2, u3))  # valid idea

For an invalid setup, a mismatch should be rejected by the type checker:

u1 = Node[int, float](...)
u2 = Node[str, complex](...)
u3 = Node[complex, int](...)

mgr = CycleManager((u1, u2, u3))  # should be rejected: float vs str

What is actually being constrained

The constraint is that for all components which are passed in, the output type TOutput of the n'th component must match the input type TInput of the (n + 1)'th component.

This is not a property of an arbitrary container; it’s a typing rule on a binary composition step between two neighboring elements. Enforcing the rule where two components meet is more natural than trying to describe the entire tuple’s shape.

This should wrap around such that the TOutput of the last component in the chain is equal to TInput of the first component in the chain.

The wrap-around check can be placed where the tuple is finally consumed. That is, the processor that takes the chain can require the first and last types in the constructed sequence to match.

A workable approach

If you are ready to accept a slightly different API, you can shift composition into a binary operator and use that to build a typed tuple progressively. The idea is to let a single block know its input/output types, and to have an accumulating tuple type that remembers only the type at the head and the tail. The binary operation enforces adjacency at each step; the processor enforces the ring closure.

Using this model, the composition rule can be expressed by overloading the binary +. You could choose another operator if you prefer. The final processor enforces that the chain’s start and end types coincide.

Code: implementation and usage

Below is a minimal implementation that enforces adjacency at composition time and wrap-around at processor construction time.

from __future__ import annotations

from typing import Any, overload


class _ChainBundle[Head, Tail](tuple[Head, *tuple[Any, ...], Tail]):
    @overload  # type: ignore[override]
    def __add__[Next](
        self, other: Node[Tail, Next], /
    ) -> _ChainBundle[Head, Next]: ...

    @overload
    def __add__[NewTail](
        self, other: _ChainBundle[Tail, NewTail], /
    ) -> _ChainBundle[Head, NewTail]: ...

    def __add__[NewTail](
        self, other: Node[Tail, NewTail] | _ChainBundle[Tail, NewTail], /
    ) -> _ChainBundle[Head, NewTail]:
        if isinstance(other, Node):
            return _ChainBundle((*self, other))
        else:
            return _ChainBundle((*self, *other))


class Node[XIn, XOut]:
    def __add__[XOut2](
        self, other: Node[XOut, XOut2], /
    ) -> _ChainBundle[XIn, XOut2]:
        return _ChainBundle((self, other))


class CycleRunner[T]:
    def __init__(self, chain: tuple[T, *tuple[Any, ...], T]): ...

With this API, composition happens with +, and types are checked at every boundary.

x1 = Node[int, str]()
x2 = Node[str, complex]()
x3 = Node[complex, int]()

ok_runner = CycleRunner[int](x1 + x2 + x3)  # OK

bad = Node[int, float]()
# The next line is rejected by the type checker due to the mismatch at '+'
# CycleRunner[int](bad + x2 + x3)
# e.g. Unsupported operand types for + ("Node[int, float]" and "Node[str, complex]")

If you adopt this API, remember to add the reflected operator variants such as __radd__ on both the single element and the accumulating tuple type.

Why this is useful

This approach enforces the local rule where it belongs, at the point of composition, and defers the global cycle rule to the final consumer. The result is a chain that either type-checks at build time or is rejected as soon as a mismatch is introduced.

Wrap-up

When you need type-level guarantees for a cyclic pipeline, express the constraint as a binary operation between neighbors and use a tuple-like accumulator that tracks only the endpoints. Let the final processor assert the wrap-around. If you stick to this pattern, the type checker will keep adjacency honest and the cycle consistent, and your chain definitions will remain both clear and verifiable.