momoa.engines

[docs] package momoa.engines

"""Engine protocol and registry for Momoa model compilation."""

from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, Protocol, runtime_checkable

from momoa.exceptions import SchemaError, UnknownEngineError


@dataclass
class EngineResult:
[docs]
    """Result of compiling a schema spec into model classes."""

    models: tuple[type, ...]

    @property
    def model(self) -> type:
[docs]
        """The primary (root) model class."""
        return self.models[-1]


@runtime_checkable
class Serializable(Protocol):
[docs]
    """Anything that can serialize itself to a JSON-compatible dict."""

    def serialize(self) -> dict[str, Any]:
[docs]
        """Return a JSON-compatible dict representation of this instance."""


class ModelEngine(Protocol):
[docs]
    """Compiles a normalised schema spec into model classes."""

    @property
    def output_format(self) -> str:
[docs]
        """Identifier: 'statham', 'pydantic', 'dataclass', etc."""
[docs]
    def compile(self, spec: dict[str, Any], *, root_name: str | None = None) -> EngineResult:
        """Returns models built from the spec. Pure: same input -> equivalent output."""
[docs]
    def context_labeller(self) -> Callable[[str], tuple[str, Any]] | None:
        """Returns a context_labeller for json_ref_dict.materialize, or None.

        Engines that need schema nodes labelled during URI materialisation (e.g.
        to derive names for anonymous nested objects) return a callable here.
        Engines that work from plain schema dicts return None.
        """


def _get_registered_engines() -> dict[str, "ModelEngine"]:
    from momoa.engines.pydantic import PydanticEngine
    from momoa.engines.statham import StathamEngine

    return {"statham": StathamEngine(), "pydantic": PydanticEngine()}


def resolve_engine(name: str) -> "ModelEngine":
[docs]
    """Look up a registered engine by name, raising UnknownEngineError for unknown names."""
    engines = _get_registered_engines()
    if name not in engines:
        raise UnknownEngineError(name, ", ".join(sorted(engines)))
    return engines[name]