"""Entity schema registry. Each business entity (CRM contact, accounting invoice, ...) declares a single `EntitySchema` so the frontend can render forms, tables and filters generically. Schemas are pure metadata — they do **not** carry business logic and they do **not** know about request context. The endpoint that serves them applies permission gates per call. """ from __future__ import annotations import re from dataclasses import dataclass, field from threading import Lock from typing import Any, ClassVar class SchemaError(RuntimeError): """Schema registration or lookup failure.""" # `module.entity` — same shape as permission codenames and event names. _DOTTED_NAME = re.compile(r"^[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)+$") # --------------------------------------------------------------------------- # Field / action / view metadata. # --------------------------------------------------------------------------- FIELD_TYPES: frozenset[str] = frozenset( { "string", "text", "int", "decimal", "bool", "date", "datetime", "uuid", "json", "enum", "fk", "m2m", } ) @dataclass(frozen=True) class FieldSpec: """Describes one entity field. `semantic_type` carries the AI-friendly meaning ("email", "money_amount", "phone") independent of storage type. `ai_hint` is free text the AI layer can use as additional context. `label_key` and `help_key` are i18n keys resolved by the frontend, never raw text. """ name: str type: str label_key: str help_key: str = "" semantic_type: str = "" ai_hint: str = "" required: bool = False read_only: bool = False default: Any = None enum: tuple[str, ...] = () fk_entity: str = "" max_length: int | None = None min_value: float | None = None max_value: float | None = None permission: str = "" # extra gate on top of entity view permission def __post_init__(self) -> None: if self.type not in FIELD_TYPES: raise SchemaError(f"Unknown field type '{self.type}' for field '{self.name}'") if self.type in {"fk", "m2m"} and not self.fk_entity: raise SchemaError(f"Field '{self.name}' of type '{self.type}' needs `fk_entity`") if self.type == "enum" and not self.enum: raise SchemaError(f"Enum field '{self.name}' needs `enum` values") @dataclass(frozen=True) class ActionSpec: """Declarative business action exposed alongside the entity. The frontend renders an action button; the backend resolves the route via the `endpoint` string. `permission` is the codename required to invoke the action. """ name: str label_key: str endpoint: str method: str = "POST" permission: str = "" confirm_key: str = "" @dataclass(frozen=True) class ViewSpec: """Describes one rendering context (list / detail / form). `fields` is the ordered field list; empty means "all fields in declaration order". `permission` lets a single view be hidden when the actor lacks the right. """ kind: str # "list" | "detail" | "form" fields: tuple[str, ...] = () permission: str = "" @dataclass(frozen=True) class EntitySchema: """Top-level descriptor for one entity.""" name: str # dotted, e.g. "crm.contact" label_key: str plural_label_key: str fields: tuple[FieldSpec, ...] views: tuple[ViewSpec, ...] = () actions: tuple[ActionSpec, ...] = () list_endpoint: str = "" detail_endpoint: str = "" view_permission: str = "" create_permission: str = "" update_permission: str = "" delete_permission: str = "" filterable_fields: tuple[str, ...] = () sortable_fields: tuple[str, ...] = () def __post_init__(self) -> None: if not _DOTTED_NAME.match(self.name): raise SchemaError(f"Entity name '{self.name}' must be dotted lowercase (module.entity)") seen: set[str] = set() for fld in self.fields: if fld.name in seen: raise SchemaError(f"Duplicate field '{fld.name}' in '{self.name}'") seen.add(fld.name) for view in self.views: for fname in view.fields: if fname not in seen: raise SchemaError( f"View '{view.kind}' of '{self.name}' references unknown field '{fname}'" ) for fname in self.filterable_fields + self.sortable_fields: if fname not in seen: raise SchemaError( f"Entity '{self.name}' lists unknown field '{fname}' as filter/sort" ) def field(self, name: str) -> FieldSpec | None: for fld in self.fields: if fld.name == name: return fld return None def serialize(self) -> dict[str, Any]: """Wire format. Keep in sync with the frontend `EntitySchema` type.""" return { "name": self.name, "label_key": self.label_key, "plural_label_key": self.plural_label_key, "list_endpoint": self.list_endpoint, "detail_endpoint": self.detail_endpoint, "permissions": { "view": self.view_permission, "create": self.create_permission, "update": self.update_permission, "delete": self.delete_permission, }, "filterable_fields": list(self.filterable_fields), "sortable_fields": list(self.sortable_fields), "fields": [ { "name": f.name, "type": f.type, "label_key": f.label_key, "help_key": f.help_key, "semantic_type": f.semantic_type, "ai_hint": f.ai_hint, "required": f.required, "read_only": f.read_only, "default": f.default, "enum": list(f.enum), "fk_entity": f.fk_entity, "max_length": f.max_length, "min_value": f.min_value, "max_value": f.max_value, "permission": f.permission, } for f in self.fields ], "views": [ {"kind": v.kind, "fields": list(v.fields), "permission": v.permission} for v in self.views ], "actions": [ { "name": a.name, "label_key": a.label_key, "endpoint": a.endpoint, "method": a.method, "permission": a.permission, "confirm_key": a.confirm_key, } for a in self.actions ], } # --------------------------------------------------------------------------- # Registry (process-wide, thread-safe, idempotent). # --------------------------------------------------------------------------- @dataclass class _Registry: entries: dict[str, EntitySchema] = field(default_factory=dict) lock: ClassVar[Lock] = Lock() _registry = _Registry() def register_entity(schema: EntitySchema) -> EntitySchema: """Register an entity schema. Idempotent for identical specs.""" with _registry.lock: existing = _registry.entries.get(schema.name) if existing is not None and existing != schema: raise SchemaError( f"Entity '{schema.name}' already registered with a different definition" ) _registry.entries[schema.name] = schema return schema def get_entity(name: str) -> EntitySchema: try: return _registry.entries[name] except KeyError as exc: raise SchemaError(f"Unknown entity '{name}'") from exc def list_entities() -> list[EntitySchema]: return list(_registry.entries.values()) def reset_registry_for_tests() -> None: with _registry.lock: _registry.entries.clear()