orchestrata_ultimate.domaingen.generator

Reference › orchestrata_ultimate › domaingen › generator

generator turns a DomainModel into domain code laid out as a screaming architecture (SPEC-105 FR-004, FR-005 and FR-006). generate_code takes the model and an output directory and writes one package per module containing entities.py, repository.py and __init__.py. They are rendered with Jinja2 from templates held in a templates directory beside the module (_env builds the environment). Output is deterministic: _topo_sort orders entities with parents before children and breaks ties by name, so output is byte-identical across runs. Writes are confined to the target root by _safe, and CodeGenError is raised on a sandbox violation. A generation-time guard also rejects any destructive SPARQL statement (such as DELETE DATA, DROP or CLEAR forms) in the generated persistence code, as required by SEC-105-003, and raises the same error. It returns the list of written paths. The input comes from reader and models; the generated repositories take a connection like stardog. The package imports it.

At a glance

Class diagram

Dependency diagram

API

Classes

CodeGenError

Bases: BaseError · Decorators: traces(spec='105', story='US-105-001', reqs=['REQ-105-048', 'REQ-105-040']) · Line 43

Raised on sandbox violations or a forbidden-SPARQL guard trip.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/domaingen/generator.py

"""DomainModel IR -> screaming-architecture domain code (SPEC-105 FR-004/005/006).

Deterministic Jinja2 codegen: one package per module with entities.py,
repository.py and __init__.py. Entities are topologically ordered (parents before
children) with a name tiebreak, so output is byte-identical across runs. Writes
are sandboxed to the target root, and a generation-time guard rejects any
destructive SPARQL (SEC-105-003).
"""

from __future__ import annotations

from pathlib import Path

from jinja2 import Environment, FileSystemLoader

from orchestrata_ultimate.core.base.error import BaseError
from orchestrata_ultimate.core.traceability import traces
from orchestrata_ultimate.domaingen.constraints import generate_shacl
from orchestrata_ultimate.domaingen.models import DomainModel, Entity

_TEMPLATE_DIR = Path(__file__).parent / "templates"
_TEMPLATES = (
    ("entities.py.j2", "entities.py"),
    ("repository.py.j2", "repository.py"),
    ("init.py.j2", "__init__.py"),
)
# SEC-105-003: generated persistence must never be destructive. Match actual
# SPARQL update statements (not the words in prose/docstrings).
_FORBIDDEN_SPARQL = (
    "DELETE DATA",
    "DELETE WHERE",
    "DELETE {",
    "DROP GRAPH",
    "DROP ALL",
    "DROP DEFAULT",
    "CLEAR GRAPH",
    "CLEAR ALL",
    "CLEAR DEFAULT",
)


@traces(spec="105", story="US-105-001", reqs=["REQ-105-048", "REQ-105-040"])
class CodeGenError(BaseError):
    """Raised on sandbox violations or a forbidden-SPARQL guard trip."""


@traces(spec="105", story="US-105-001", reqs=["REQ-105-006"])
def _env() -> Environment:
    return Environment(
        loader=FileSystemLoader(str(_TEMPLATE_DIR)),
        trim_blocks=True,
        lstrip_blocks=True,
        keep_trailing_newline=True,
        autoescape=False,
    )


@traces(spec="105", story="US-105-001", reqs=["REQ-105-006"])
def _topo_sort(entities: list[Entity]) -> list[Entity]:
    """Parents before children; deterministic (sorted by name, stable)."""
    by_name = {e.name: e for e in entities}
    ordered: list[Entity] = []
    seen: set[str] = set()

    @traces(spec="105", story="US-105-001", reqs=["REQ-105-006"])
    def visit(entity: Entity) -> None:
        if entity.name in seen:
            return
        seen.add(entity.name)
        if entity.parent and entity.parent in by_name:
            visit(by_name[entity.parent])
        ordered.append(entity)

    for entity in sorted(entities, key=lambda e: e.name):
        visit(entity)
    return ordered


@traces(
    spec="105",
    story="US-105-001",
    reqs=[
        "REQ-105-005",
        "REQ-105-006",
        "REQ-105-008",
        "REQ-105-040",
        "REQ-105-043",
    ],
)
def generate_code(model: DomainModel, out_dir: str | Path) -> list[Path]:
    """Generate the domain model from the IR; return the written file paths."""
    root = Path(out_dir).resolve()
    env = _env()
    written: list[Path] = []

    for module in model.modules:
        entities = _topo_sort(module.entities)
        context = {
            "module_name": module.name,
            "entities": entities,
            "namespace": model.namespace,
        }
        module_dir = _safe(root, module.name)
        module_dir.mkdir(parents=True, exist_ok=True)
        for template_name, filename in _TEMPLATES:
            content = env.get_template(template_name).render(**context)
            for bad in _FORBIDDEN_SPARQL:
                if bad in content:
                    raise CodeGenError(
                        f"generated {filename} contains forbidden SPARQL {bad!r}"
                    )
            target = _safe(root, f"{module.name}/{filename}")
            target.write_text(content, encoding="utf-8")
            written.append(target)

    # Ontology-wide SHACL constraints for Stardog ICV (FR-008).
    constraints = _safe(root, "constraints.ttl")
    constraints.write_text(generate_shacl(model), encoding="utf-8")
    written.append(constraints)

    return sorted(written)


@traces(spec="105", story="US-105-001", reqs=["REQ-105-048"])
def _safe(root: Path, rel: str) -> Path:
    resolved = (root / rel).resolve()
    if resolved != root and root not in resolved.parents:
        raise CodeGenError(f"path escapes target root: {rel}")
    return resolved