"""# Summary Example pipeline family — concrete stages, the wired root, and its registration. The concrete subclass re-annotates each slot with a stage implementation; the DI container builds the whole tree. Register the family so the shared registry can resolve it by ``(kind, variant, mode)``. """ from pathlib import Path from typing import Any from pipelines.example.abc import ( ExamplePipelinePort, IngestPort, NormalizePort, ReadPort, ) from pipelines.shared.loader import build_pipeline from pipelines.shared.ports import FinalizePort from pipelines.shared.registry import register class TextIngest(IngestPort): """Materialize any source as text.""" def materialize(self, source: Any) -> str: return source if isinstance(source, str) else str(source) class LineReader(ReadPort): """Read one record per non-empty line.""" def read(self, content: str) -> list[dict[str, Any]]: return [{"line": line} for line in content.splitlines() if line.strip()] class PassthroughNormalize(NormalizePort): """Return records unchanged (replace with real normalization).""" def build(self, records: list[dict[str, Any]]) -> list[dict[str, Any]]: return records class NoopFinalize(FinalizePort): """Drop records (replace with a service-backed finalize stage).""" async def finalize(self, records: list[dict[str, Any]]) -> list[Any]: return records class ExamplePipeline(ExamplePipelinePort): """Concrete example family — the dependency decisions live here.""" ingest: TextIngest reader: LineReader normalize: PassthroughNormalize finalize: NoopFinalize async def run(self, source: Any) -> list[dict[str, Any]]: content = self.ingest.materialize(source) records = self.reader.read(content) return self.normalize.build(records) def get_example_pipeline() -> ExamplePipeline: """Wire and return the example pipeline through the shared loader.""" resources = Path(__file__).parent / "resources" return build_pipeline(ExamplePipeline, resources_dir=resources) register("example", "default", ExamplePipeline)