Multi Stage Pipeline With Fail Fast Ordering

from typing import Any, Callable, List, Optional, Sequence, Tuple


class StageError(RuntimeError):
    """Raised when a pipeline stage fails or breaks the record invariant."""

    def __init__(self, stage: str, index: Optional[int], message: str) -> None:
        self.stage = stage
        self.index = index
        location = stage if index is None else "%s[%d]" % (stage, index)
        super().__init__("%s: %s" % (location, message))


def run_pipeline(
    records: Sequence[Any],
    stages: Sequence[Tuple[str, Callable[[Any], Any]]],
) -> List[Any]:
    """Run named stages in order; fail fast with stage and index context."""
    if not records:
        raise StageError("pipeline", None, "no input records")

    current: List[Any] = list(records)
    for name, stage in stages:
        produced: List[Any] = []
        for index, record in enumerate(current):
            try:
                produced.append(stage(record))
            except Exception as exc:
                raise StageError(name, index, str(exc)) from exc
        if len(produced) != len(current):
            raise StageError(
                name,
                None,
                "record count changed from %d to %d" % (len(current), len(produced)),
            )
        current = produced
    return current