diff --git a/standards/src/templates/app/domain/item/ports/converters.py b/standards/src/templates/app/domain/item/ports/converters.py new file mode 100644 index 0000000..f386df4 --- /dev/null +++ b/standards/src/templates/app/domain/item/ports/converters.py @@ -0,0 +1,18 @@ +"""# Summary + +Optional converters for the item slice — pure functions mapping between domain +types. Add a converter here when a service must reshape data; keep it pure. +""" +from domain.item.models import Item + + +def to_item(data: dict) -> Item: + """Build an :class:`Item` from a raw mapping. + + Args: + data (dict): Field mapping matching the item model. + + Returns: + Item: The constructed item model. + """ + return Item(**data) diff --git a/standards/src/templates/app/pipelines/example/__init__.py b/standards/src/templates/app/pipelines/example/__init__.py new file mode 100644 index 0000000..196d174 --- /dev/null +++ b/standards/src/templates/app/pipelines/example/__init__.py @@ -0,0 +1,7 @@ +"""# Summary + +Example pipeline family — a reusable skeleton, not an app-specific pipeline. + +Replace ``Example*`` names and the three stages with your family's stages. The +shape (abc → impl → pipeline.yaml → resources/) stays the same. +""" diff --git a/standards/src/templates/app/pipelines/example/abc.py b/standards/src/templates/app/pipelines/example/abc.py new file mode 100644 index 0000000..ebbc570 --- /dev/null +++ b/standards/src/templates/app/pipelines/example/abc.py @@ -0,0 +1,53 @@ +"""# Summary + +Example pipeline family — port contracts (stage interfaces + root interface). + +Ports are declared as a DI tree: the root's annotated slots *are* the dependency +decisions. The concrete subclass in ``impl.py`` pins the stage types. +""" +from abc import ABC, abstractmethod +from typing import Any + +from pipelines.shared.ports import FinalizePort + + +class IngestPort(ABC): + """Obtain the raw input as text/bytes.""" + + @abstractmethod + def materialize(self, source: Any) -> str: + """Return the raw input materialized as text.""" + + +class ReadPort(ABC): + """Parse materialized input into records.""" + + @abstractmethod + def read(self, content: str) -> list[dict[str, Any]]: + """Return parsed records.""" + + +class NormalizePort(ABC): + """Normalize raw records into the target shape.""" + + @abstractmethod + def build(self, records: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Return normalized records.""" + + +class ExamplePipelinePort(ABC): + """Root of the example family. + + Notes: + The slot annotations below are contracts. A concrete subclass re-annotates + them with stage implementations; the DI container instantiates the tree. + """ + + ingest: IngestPort + reader: ReadPort + normalize: NormalizePort + finalize: FinalizePort + + @abstractmethod + async def run(self, source: Any) -> list[dict[str, Any]]: + """Run the full stage chain and return the normalized records.""" diff --git a/standards/src/templates/app/pipelines/example/impl.py b/standards/src/templates/app/pipelines/example/impl.py new file mode 100644 index 0000000..d78c30e --- /dev/null +++ b/standards/src/templates/app/pipelines/example/impl.py @@ -0,0 +1,71 @@ +"""# 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) diff --git a/standards/src/templates/app/pipelines/example/pipeline.yaml b/standards/src/templates/app/pipelines/example/pipeline.yaml new file mode 100644 index 0000000..0ebfe2d --- /dev/null +++ b/standards/src/templates/app/pipelines/example/pipeline.yaml @@ -0,0 +1,18 @@ +# Graph spec for the example family: nodes, edges, and the root. +# Mirrors the port tree in abc.py. Replace with your family's graph. +schema: ExampleRecord +nodes: + ingest: + port: IngestPort + reader: + port: ReadPort + normalize: + port: NormalizePort + finalize: + port: FinalizePort +edges: + - ingest -> reader + - reader -> normalize + - normalize -> finalize +roots: + default: ExamplePipeline diff --git a/standards/src/templates/app/pipelines/example/resources/patterns/example.yaml b/standards/src/templates/app/pipelines/example/resources/patterns/example.yaml new file mode 100644 index 0000000..0074eff --- /dev/null +++ b/standards/src/templates/app/pipelines/example/resources/patterns/example.yaml @@ -0,0 +1,4 @@ +# Pattern / rule fragments for the example family. +# The loader merges every patterns/*.yaml into one registry by key. +example: + enabled: true diff --git a/standards/src/templates/app/pipelines/example/resources/pipeline.yaml b/standards/src/templates/app/pipelines/example/resources/pipeline.yaml new file mode 100644 index 0000000..8a670be --- /dev/null +++ b/standards/src/templates/app/pipelines/example/resources/pipeline.yaml @@ -0,0 +1,5 @@ +# Runtime spec consumed by the shared loader. +kind: example +variant: default +mode: default +patterns: [] diff --git a/standards/src/templates/app/pipelines/shared/__init__.py b/standards/src/templates/app/pipelines/shared/__init__.py new file mode 100644 index 0000000..8aef39c --- /dev/null +++ b/standards/src/templates/app/pipelines/shared/__init__.py @@ -0,0 +1,5 @@ +"""# Summary + +Shared pipeline scaffolding — ports, loader, and registry reused by every +pipeline family. See standards "06_pipeline_scaffold". +""" diff --git a/standards/src/templates/app/pipelines/shared/loader.py b/standards/src/templates/app/pipelines/shared/loader.py new file mode 100644 index 0000000..10a3c2b --- /dev/null +++ b/standards/src/templates/app/pipelines/shared/loader.py @@ -0,0 +1,54 @@ +"""# Summary + +Pipeline loader — wire a family root through the DI library and attach the +merged pattern registry from ``resources/patterns/*.yaml``. + +All families share this loader so wiring is never hand-rolled. +""" +from pathlib import Path +from typing import Any + +import yaml + +from hexa import build + + +def load_patterns(resources_dir: str | Path) -> dict[str, Any]: + """Merge every pattern fragment under ``resources/patterns`` into one map. + + Args: + resources_dir (str | Path): Family resources directory. + + Returns: + dict[str, Any]: Merged pattern/rule fragments keyed by fragment name. + """ + registry: dict[str, Any] = {} + patterns = Path(resources_dir) / "patterns" + if patterns.is_dir(): + for path in sorted(patterns.glob("*.yaml")): + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + registry.update(data) + return registry + + +def build_pipeline( + root_cls, + *, + resources_dir: str | Path = "resources", + instances: dict[str, Any] | None = None, + config: dict[str, Any] | None = None, +): + """Wire ``root_cls`` and inject the merged pattern registry. + + Args: + root_cls: The pipeline root class (a port tree). + resources_dir (str | Path): Directory holding ``patterns/*.yaml``. + instances (dict[str, Any] | None): Runtime objects injected by slot name. + config (dict[str, Any] | None): Config values propagated by field name. + + Returns: + The fully wired pipeline root. + """ + merged_instances = dict(instances or {}) + merged_instances.setdefault("registry", load_patterns(resources_dir)) + return build(root_cls, instances=merged_instances, config=config or {}) diff --git a/standards/src/templates/app/pipelines/shared/ports.py b/standards/src/templates/app/pipelines/shared/ports.py new file mode 100644 index 0000000..ec15d8e --- /dev/null +++ b/standards/src/templates/app/pipelines/shared/ports.py @@ -0,0 +1,25 @@ +"""# Summary + +Shared pipeline ports — the contracts every pipeline family reuses. + +A port is an abstract stage contract. Concrete families pin their implementations +on a root's annotated slots; the DI library instantiates the tree. +""" +from abc import ABC, abstractmethod +from typing import Any + + +class FinalizePort(ABC): + """Terminal stage — turn normalized records into persisted resources.""" + + @abstractmethod + async def finalize(self, records: list[dict[str, Any]]) -> list[Any]: + """Persist records and return the created resources.""" + + +class RegistryPort(ABC): + """Resolve a pattern / rule fragment by key.""" + + @abstractmethod + def resolve(self, key: str) -> dict[str, Any]: + """Return the fragment registered under ``key``.""" diff --git a/standards/src/templates/app/pipelines/shared/registry.py b/standards/src/templates/app/pipelines/shared/registry.py new file mode 100644 index 0000000..d5de929 --- /dev/null +++ b/standards/src/templates/app/pipelines/shared/registry.py @@ -0,0 +1,29 @@ +"""# Summary + +Family registry — resolve a ``(kind, variant, mode)`` triple to a root class. + +A single lookup replaces hand-rolled ``if/else`` family selection. +""" +from typing import Dict, Type + +# Populated by ``register`` as families import. +_FAMILIES: Dict[str, Dict[str, Dict[str, Type]]] = {} + + +def register(kind: str, variant: str, root_cls: Type, mode: str = "default") -> Type: + """Register ``root_cls`` for a ``(kind, variant, mode)`` triple. + + Returns: + Type: The registered class (so it can be used as a decorator). + """ + _FAMILIES.setdefault(kind, {}).setdefault(variant, {})[mode] = root_cls + return root_cls + + +def resolve_root(kind: str, variant: str, mode: str = "default") -> Type: + """Return the root class for a ``(kind, variant, mode)`` triple. + + Raises: + KeyError: When no family matches the triple. + """ + return _FAMILIES[kind][variant][mode] diff --git a/standards/src/templates/app/pyproject.toml b/standards/src/templates/app/pyproject.toml index 9d44827..48288f1 100644 --- a/standards/src/templates/app/pyproject.toml +++ b/standards/src/templates/app/pyproject.toml @@ -12,6 +12,7 @@ dependencies = [ "pydantic>=2", "motor", "python-dotenv", + "pyyaml", "openapi-first", "mongo-ops", "hexa", @@ -29,7 +30,7 @@ dev = [ [tool.setuptools.packages.find] where = ["."] -include = ["domain*", "infra*", "interfaces*", "pipelines*"] +include = ["domain*", "infra*", "interfaces*", "pipelines*", "scripts*"] [tool.pytest.ini_options] asyncio_mode = "auto" diff --git a/standards/src/templates/app/scripts/__init__.py b/standards/src/templates/app/scripts/__init__.py new file mode 100644 index 0000000..43e0740 --- /dev/null +++ b/standards/src/templates/app/scripts/__init__.py @@ -0,0 +1,7 @@ +"""# Summary + +Operator scripts — run from the project root as ``python -m scripts.``. + +They use the composition root (:func:`init.init` / :func:`init.shutdown`) so they +share the exact wiring the API and worker use. +""" diff --git a/standards/src/templates/app/scripts/drop.py b/standards/src/templates/app/scripts/drop.py new file mode 100644 index 0000000..042064a --- /dev/null +++ b/standards/src/templates/app/scripts/drop.py @@ -0,0 +1,32 @@ +"""# Summary + +Operator script — drop the collections owned by each service for a clean slate. + +Run from the project root: + + python -m scripts.drop +""" +import asyncio + +from init import init, shutdown + + +async def main() -> None: + """Delete every service collection, best-effort, then shut down.""" + resources = await init() + try: + dropped = 0 + for name, svc in resources.items(): + if not name.endswith("_svc"): + continue + delete_many = getattr(svc.repo, "delete_many", None) + if delete_many is not None: + await delete_many({}) + dropped += 1 + print(f"Dropped {dropped} service collection(s).") + finally: + await shutdown() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/standards/src/templates/app/scripts/seed.py b/standards/src/templates/app/scripts/seed.py new file mode 100644 index 0000000..2a2ad0f --- /dev/null +++ b/standards/src/templates/app/scripts/seed.py @@ -0,0 +1,26 @@ +"""# Summary + +Operator script — seed demo data through the composition root. + +Run from the project root: + + python -m scripts.seed +""" +import asyncio + +from init import init, shutdown + + +async def main() -> None: + """Seed one demo item and shut down cleanly.""" + resources = await init() + try: + item_svc = resources["item_svc"] + await item_svc.create({"name": "seed", "quantity": 1}) + print("Seeded one item.") + finally: + await shutdown() + + +if __name__ == "__main__": + asyncio.run(main())