rose-ash/shared/sx/relations.py

"""
Relation registry — declarative entity relationship definitions.

Relations are defined as s-expressions using ``defrelation`` and stored
in a global registry.  All services load the same definitions at startup
via ``load_relation_registry()``.

No evaluator dependency — defrelation forms are parsed directly from the
AST since they're just structured data (keyword args → RelationDef).
"""

from __future__ import annotations

from shared.sx.types import Keyword, RelationDef, Symbol


# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------

_RELATION_REGISTRY: dict[str, RelationDef] = {}


def register_relation(defn: RelationDef) -> None:
    """Add a RelationDef to the global registry."""
    _RELATION_REGISTRY[defn.name] = defn


def get_relation(name: str) -> RelationDef | None:
    """Look up a relation by name (e.g. ``"page->market"``)."""
    return _RELATION_REGISTRY.get(name)


def relations_from(entity_type: str) -> list[RelationDef]:
    """All relations where *entity_type* is the ``from`` side."""
    return [d for d in _RELATION_REGISTRY.values() if d.from_type == entity_type]


def relations_to(entity_type: str) -> list[RelationDef]:
    """All relations where *entity_type* is the ``to`` side."""
    return [d for d in _RELATION_REGISTRY.values() if d.to_type == entity_type]


def all_relations() -> list[RelationDef]:
    """Return all registered relations."""
    return list(_RELATION_REGISTRY.values())


def clear_registry() -> None:
    """Clear all registered relations (for testing)."""
    _RELATION_REGISTRY.clear()


# ---------------------------------------------------------------------------
# defrelation parsing — direct AST walk, no evaluator needed
# ---------------------------------------------------------------------------

_VALID_CARDINALITIES = {"one-to-one", "one-to-many", "many-to-many"}
_VALID_NAV = {"submenu", "tab", "badge", "inline", "hidden"}


class RelationError(Exception):
    """Error parsing a defrelation form."""
    pass


def _parse_defrelation(expr: list) -> RelationDef:
    """Parse a (defrelation :name :key val ...) AST into a RelationDef."""
    if len(expr) < 2:
        raise RelationError("defrelation requires a name")

    name_kw = expr[1]
    if not isinstance(name_kw, Keyword):
        raise RelationError(
            f"defrelation name must be a keyword, got {type(name_kw).__name__}"
        )
    rel_name = name_kw.name

    # Parse keyword args
    kwargs: dict[str, str | None] = {}
    i = 2
    while i < len(expr):
        key = expr[i]
        if isinstance(key, Keyword):
            if i + 1 < len(expr):
                val = expr[i + 1]
                kwargs[key.name] = val.name if isinstance(val, Keyword) else val
                i += 2
            else:
                kwargs[key.name] = None
                i += 1
        else:
            i += 1

    for field in ("from", "to", "cardinality"):
        if field not in kwargs:
            raise RelationError(
                f"defrelation {rel_name} missing required :{field}"
            )

    card = kwargs["cardinality"]
    if card not in _VALID_CARDINALITIES:
        raise RelationError(
            f"defrelation {rel_name}: invalid cardinality {card!r}, "
            f"expected one of {_VALID_CARDINALITIES}"
        )

    nav = kwargs.get("nav", "hidden")
    if nav not in _VALID_NAV:
        raise RelationError(
            f"defrelation {rel_name}: invalid nav {nav!r}, "
            f"expected one of {_VALID_NAV}"
        )

    return RelationDef(
        name=rel_name,
        from_type=kwargs["from"],
        to_type=kwargs["to"],
        cardinality=card,
        inverse=kwargs.get("inverse"),
        nav=nav,
        nav_icon=kwargs.get("nav-icon"),
        nav_label=kwargs.get("nav-label"),
    )


def evaluate_defrelation(expr: list) -> RelationDef:
    """Parse a defrelation form, register it, and return the RelationDef.

    Also handles (begin (defrelation ...) ...) wrappers.
    """
    if not isinstance(expr, list) or not expr:
        raise RelationError(f"Expected list expression, got {type(expr).__name__}")

    head = expr[0]
    if isinstance(head, Symbol) and head.name == "begin":
        result = None
        for child in expr[1:]:
            result = evaluate_defrelation(child)
        return result

    if not (isinstance(head, Symbol) and head.name == "defrelation"):
        raise RelationError(f"Expected defrelation, got {head}")

    defn = _parse_defrelation(expr)
    register_relation(defn)
    return defn


# ---------------------------------------------------------------------------
# Built-in relation definitions (s-expression source)
# ---------------------------------------------------------------------------

_BUILTIN_RELATIONS = '''
(begin

  (defrelation :page->market
    :from "page"
    :to "market"
    :cardinality :one-to-many
    :inverse :market->page
    :nav :submenu
    :nav-icon "fa fa-shopping-bag"
    :nav-label "markets")

  (defrelation :page->calendar
    :from "page"
    :to "calendar"
    :cardinality :one-to-many
    :inverse :calendar->page
    :nav :submenu
    :nav-icon "fa fa-calendar"
    :nav-label "calendars")

  (defrelation :post->calendar_entry
    :from "post"
    :to "calendar_entry"
    :cardinality :many-to-many
    :inverse :calendar_entry->post
    :nav :inline
    :nav-icon "fa fa-file-alt"
    :nav-label "events")

  (defrelation :page->menu_node
    :from "page"
    :to "menu_node"
    :cardinality :one-to-one
    :nav :hidden)

)
'''


def load_relation_registry() -> None:
    """Parse built-in defrelation s-expressions and populate the registry."""
    from shared.sx.parser import parse

    tree = parse(_BUILTIN_RELATIONS)
    evaluate_defrelation(tree)