Vishesh 'ironeagle' Bangotra
6c0d859a1e
- F11 PopulateRule field_name/collection_name corrections - F12 service adaptor update handles PopulatingRepository - F13 nested-references section; F14 authless /users/me guidance - F15 template ruff-clean; F16 stop vendoring .ruff_cache, ship .gitignore - document --template-style in build_recipe/spec_first_routing; refresh collected openapi-first + standards bundle (12 modules, 43 components) - add trial.md and trial_findings.md (F1-F16, re-trial A/B evidence)
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
"""# 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 {})
|