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