docs(standards): add reusable Tier-B template scaffolds (pipeline family, shared loader/registry, converters, scripts)

This commit is contained in:
2026-10-12 21:08:19 +05:30
parent 99fac53c2b
commit 52b4d76d93
15 changed files with 356 additions and 1 deletions

View File

@@ -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)

View File

@@ -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.
"""

View File

@@ -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."""

View File

@@ -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)

View File

@@ -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

View File

@@ -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

View File

@@ -0,0 +1,5 @@
# Runtime spec consumed by the shared loader.
kind: example
variant: default
mode: default
patterns: []

View File

@@ -0,0 +1,5 @@
"""# Summary
Shared pipeline scaffolding — ports, loader, and registry reused by every
pipeline family. See standards "06_pipeline_scaffold".
"""

View File

@@ -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 {})

View File

@@ -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``."""

View File

@@ -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]

View File

@@ -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"

View File

@@ -0,0 +1,7 @@
"""# Summary
Operator scripts — run from the project root as ``python -m scripts.<name>``.
They use the composition root (:func:`init.init` / :func:`init.shutdown`) so they
share the exact wiring the API and worker use.
"""

View File

@@ -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())

View File

@@ -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())