art-dag/mono
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement flexible entity relation system (Phases A–E)

Browse Source 
Declarative relation registry via defrelation s-expressions with
cardinality enforcement (one-to-one, one-to-many, many-to-many),
registry-aware relate/unrelate/can-relate API endpoints, generic
container-nav fragment, and relation-driven UI components.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
giles
2026-02-28 08:35:17 +00:00
parent 6f1d5bac3c
commit a0a0f5ebc2
17 changed files with 928 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
relations/alembic/versions/0002_add_relation_type_and_metadata.py Normal file
View File

@@ -0,0 +1,52 @@
"""Add relation_type and metadata columns to container_relations
Revision ID: relations_0002
Revises: relations_0001
Create Date: 2026-02-28
"""
import sqlalchemy as sa
from alembic import op
revision = "relations_0002"
down_revision = "relations_0001"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column(
"container_relations",
sa.Column("relation_type", sa.String(64), nullable=True),
)
op.add_column(
"container_relations",
sa.Column("metadata", sa.JSON(), nullable=True),
)
op.create_index(
"ix_container_relations_relation_type",
"container_relations",
["relation_type", "parent_type", "parent_id"],
)
# Backfill relation_type for existing rows based on parent/child type pairs
op.execute("""
UPDATE container_relations
SET relation_type = CASE
WHEN parent_type = 'page' AND child_type = 'market'
THEN 'page->market'
WHEN parent_type = 'page' AND child_type = 'calendar'
THEN 'page->calendar'
WHEN parent_type = 'page' AND child_type = 'menu_node'
THEN 'page->menu_node'
END
WHERE relation_type IS NULL
AND parent_type = 'page'
AND child_type IN ('market', 'calendar', 'menu_node')
""")
def downgrade() -> None:
op.drop_index("ix_container_relations_relation_type", "container_relations")
op.drop_column("container_relations", "metadata")
op.drop_column("container_relations", "relation_type")

3
relations/app.py
View File

@@ -3,7 +3,7 @@ import path_setup # noqa: F401
from shared.infrastructure.factory import create_base_app
from bp import register_actions, register_data
from bp import register_actions, register_data, register_fragments
from services import register_domain_services
@@ -15,6 +15,7 @@ def create_app() -> "Quart":
app.register_blueprint(register_actions())
app.register_blueprint(register_data())
app.register_blueprint(register_fragments())
return app

1
relations/bp/__init__.py
View File

@@ -1,2 +1,3 @@
from .data.routes import register as register_data
from .actions.routes import register as register_actions
from .fragments.routes import register as register_fragments

116
relations/bp/actions/routes.py
View File

@@ -46,6 +46,8 @@ def register() -> Blueprint:
child_id=data["child_id"],
label=data.get("label"),
sort_order=data.get("sort_order"),
relation_type=data.get("relation_type"),
metadata=data.get("metadata"),
)
return {
"id": rel.id,
@@ -54,6 +56,7 @@ def register() -> Blueprint:
"child_type": rel.child_type,
"child_id": rel.child_id,
"sort_order": rel.sort_order,
"relation_type": rel.relation_type,
}
_handlers["attach-child"] = _attach_child
@@ -70,9 +73,122 @@ def register() -> Blueprint:
parent_id=data["parent_id"],
child_type=data["child_type"],
child_id=data["child_id"],
relation_type=data.get("relation_type"),
)
return {"deleted": deleted}
_handlers["detach-child"] = _detach_child
# --- relate (registry-aware) ---
async def _relate():
"""Create a typed relation with registry validation and cardinality enforcement."""
from shared.services.relationships import attach_child, get_children
from shared.sexp.relations import get_relation
data = await request.get_json(force=True)
rel_type = data.get("relation_type")
if not rel_type:
return {"error": "relation_type is required"}, 400
defn = get_relation(rel_type)
if defn is None:
return {"error": f"unknown relation_type: {rel_type}"}, 400
from_id = data["from_id"]
to_id = data["to_id"]
# Cardinality enforcement
if defn.cardinality == "one-to-one":
existing = await get_children(
g.s,
parent_type=defn.from_type,
parent_id=from_id,
child_type=defn.to_type,
relation_type=rel_type,
)
if existing:
return {"error": "one-to-one relation already exists", "existing_id": existing[0].child_id}, 409
rel = await attach_child(
g.s,
parent_type=defn.from_type,
parent_id=from_id,
child_type=defn.to_type,
child_id=to_id,
label=data.get("label"),
sort_order=data.get("sort_order"),
relation_type=rel_type,
metadata=data.get("metadata"),
)
return {
"id": rel.id,
"relation_type": rel.relation_type,
"parent_type": rel.parent_type,
"parent_id": rel.parent_id,
"child_type": rel.child_type,
"child_id": rel.child_id,
"sort_order": rel.sort_order,
}
_handlers["relate"] = _relate
# --- unrelate (registry-aware) ---
async def _unrelate():
"""Remove a typed relation with registry validation."""
from shared.services.relationships import detach_child
from shared.sexp.relations import get_relation
data = await request.get_json(force=True)
rel_type = data.get("relation_type")
if not rel_type:
return {"error": "relation_type is required"}, 400
defn = get_relation(rel_type)
if defn is None:
return {"error": f"unknown relation_type: {rel_type}"}, 400
deleted = await detach_child(
g.s,
parent_type=defn.from_type,
parent_id=data["from_id"],
child_type=defn.to_type,
child_id=data["to_id"],
relation_type=rel_type,
)
return {"deleted": deleted}
_handlers["unrelate"] = _unrelate
# --- can-relate (pre-flight check) ---
async def _can_relate():
"""Check if a relation can be created (cardinality, registry validation)."""
from shared.services.relationships import get_children
from shared.sexp.relations import get_relation
data = await request.get_json(force=True)
rel_type = data.get("relation_type")
if not rel_type:
return {"error": "relation_type is required"}, 400
defn = get_relation(rel_type)
if defn is None:
return {"allowed": False, "reason": f"unknown relation_type: {rel_type}"}
from_id = data["from_id"]
if defn.cardinality == "one-to-one":
existing = await get_children(
g.s,
parent_type=defn.from_type,
parent_id=from_id,
child_type=defn.to_type,
relation_type=rel_type,
)
if existing:
return {"allowed": False, "reason": "one-to-one relation already exists"}
return {"allowed": True}
_handlers["can-relate"] = _can_relate
return bp

47
relations/bp/data/routes.py
View File

@@ -35,22 +35,43 @@ def register() -> Blueprint:
parent_type = request.args.get("parent_type", "")
parent_id = request.args.get("parent_id", type=int)
child_type = request.args.get("child_type")
relation_type = request.args.get("relation_type")
if not parent_type or parent_id is None:
return []
rels = await get_children(g.s, parent_type, parent_id, child_type)
return [
{
"id": r.id,
"parent_type": r.parent_type,
"parent_id": r.parent_id,
"child_type": r.child_type,
"child_id": r.child_id,
"sort_order": r.sort_order,
"label": r.label,
}
for r in rels
]
rels = await get_children(g.s, parent_type, parent_id, child_type, relation_type=relation_type)
return [_serialize_rel(r) for r in rels]
_handlers["get-children"] = _get_children
# --- get-parents ---
async def _get_parents():
"""Return ContainerRelation parents for a child."""
from shared.services.relationships import get_parents
child_type = request.args.get("child_type", "")
child_id = request.args.get("child_id", type=int)
parent_type = request.args.get("parent_type")
relation_type = request.args.get("relation_type")
if not child_type or child_id is None:
return []
rels = await get_parents(g.s, child_type, child_id, parent_type, relation_type=relation_type)
return [_serialize_rel(r) for r in rels]
_handlers["get-parents"] = _get_parents
return bp
def _serialize_rel(r):
"""Serialize a ContainerRelation to a dict."""
return {
"id": r.id,
"parent_type": r.parent_type,
"parent_id": r.parent_id,
"child_type": r.child_type,
"child_id": r.child_id,
"sort_order": r.sort_order,
"label": r.label,
"relation_type": r.relation_type,
"metadata": r.metadata_,
}

0
relations/bp/fragments/__init__.py Normal file
View File

88
relations/bp/fragments/routes.py Normal file
View File

@@ -0,0 +1,88 @@
"""Relations app fragment endpoints.
Generic container-nav fragment that renders navigation items for all
related entities, driven by the relation registry.
"""
from __future__ import annotations
from quart import Blueprint, Response, g, request
from shared.infrastructure.fragments import FRAGMENT_HEADER
def register():
bp = Blueprint("fragments", __name__, url_prefix="/internal/fragments")
_handlers: dict[str, object] = {}
@bp.before_request
async def _require_fragment_header():
if not request.headers.get(FRAGMENT_HEADER):
return Response("", status=403)
@bp.get("/<fragment_type>")
async def get_fragment(fragment_type: str):
handler = _handlers.get(fragment_type)
if handler is None:
return Response("", status=200, content_type="text/html")
html = await handler()
return Response(html, status=200, content_type="text/html")
# --- generic container-nav fragment ----------------------------------------
async def _container_nav_handler():
"""Render nav items for all visible relations of a container entity.
Query params:
container_type: entity type (e.g. "page")
container_id: entity id
post_slug: used for URL construction
exclude: comma-separated relation types to skip
"""
from shared.sexp.jinja_bridge import sexp as render_sexp
from shared.sexp.relations import relations_from
from shared.services.relationships import get_children
container_type = request.args.get("container_type", "page")
container_id = int(request.args.get("container_id", 0))
post_slug = request.args.get("post_slug", "")
exclude_raw = request.args.get("exclude", "")
exclude = set(exclude_raw.split(",")) if exclude_raw else set()
nav_defs = [
d for d in relations_from(container_type)
if d.nav != "hidden" and d.name not in exclude
]
if not nav_defs:
return ""
parts = []
for defn in nav_defs:
children = await get_children(
g.s,
parent_type=container_type,
parent_id=container_id,
child_type=defn.to_type,
relation_type=defn.name,
)
for child in children:
slug = (child.metadata_ or {}).get("slug", "")
href = f"/{post_slug}/{slug}/" if post_slug else f"/{slug}/"
parts.append(render_sexp(
'(~relation-nav :href href :name name :icon icon :nav-class nav-class :relation-type relation-type)',
href=href,
name=child.label or "",
icon=defn.nav_icon or "",
**{
"nav-class": "",
"relation-type": defn.name,
},
))
return "\n".join(parts)
_handlers["container-nav"] = _container_nav_handler
return bp

2
shared/infrastructure/factory.py
View File

@@ -30,6 +30,7 @@ from .jinja_setup import setup_jinja
from .user_loader import load_current_user
from shared.sexp.jinja_bridge import setup_sexp_bridge
from shared.sexp.components import load_shared_components
from shared.sexp.relations import load_relation_registry
# Async init of config (runs once at import)
@@ -108,6 +109,7 @@ def create_base_app(
setup_jinja(app)
setup_sexp_bridge(app)
load_shared_components()
load_relation_registry()
errors(app)
# Auto-register OAuth client blueprint for non-account apps

9
shared/models/container_relation.py
View File

@@ -1,7 +1,7 @@
from datetime import datetime
from typing import Optional
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import Integer, String, DateTime, Index, UniqueConstraint, func
from sqlalchemy import Integer, String, DateTime, Index, JSON, UniqueConstraint, func
from shared.db.base import Base
@@ -15,6 +15,10 @@ class ContainerRelation(Base):
),
Index("ix_container_relations_parent", "parent_type", "parent_id"),
Index("ix_container_relations_child", "child_type", "child_id"),
Index(
"ix_container_relations_relation_type",
"relation_type", "parent_type", "parent_id",
),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
@@ -24,6 +28,9 @@ class ContainerRelation(Base):
child_type: Mapped[str] = mapped_column(String(32), nullable=False)
child_id: Mapped[int] = mapped_column(Integer, nullable=False)
relation_type: Mapped[Optional[str]] = mapped_column(String(64), nullable=True, index=False)
metadata_: Mapped[Optional[dict]] = mapped_column("metadata", JSON, nullable=True)
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
label: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)

72
shared/services/relationships.py
View File

@@ -15,6 +15,8 @@ async def attach_child(
child_id: int,
label: str | None = None,
sort_order: int | None = None,
relation_type: str | None = None,
metadata: dict | None = None,
) -> ContainerRelation:
"""
Create a ContainerRelation and emit container.child_attached event.
@@ -39,6 +41,10 @@ async def attach_child(
existing.sort_order = sort_order
if label is not None:
existing.label = label
if relation_type is not None:
existing.relation_type = relation_type
if metadata is not None:
existing.metadata_ = metadata
await session.flush()
await emit_activity(
session,
@@ -50,12 +56,25 @@ async def attach_child(
"parent_id": parent_id,
"child_type": child_type,
"child_id": child_id,
**({"relation_type": relation_type} if relation_type else {}),
},
source_type="container_relation",
source_id=existing.id,
)
return existing
# Already attached and active — no-op
# Already attached and active — update mutable fields if provided
changed = False
if relation_type is not None and existing.relation_type != relation_type:
existing.relation_type = relation_type
changed = True
if metadata is not None and existing.metadata_ != metadata:
existing.metadata_ = metadata
changed = True
if label is not None and existing.label != label:
existing.label = label
changed = True
if changed:
await session.flush()
return existing
if sort_order is None:
@@ -75,6 +94,8 @@ async def attach_child(
child_id=child_id,
label=label,
sort_order=sort_order,
relation_type=relation_type,
metadata_=metadata,
)
session.add(rel)
await session.flush()
@@ -89,6 +110,7 @@ async def attach_child(
"parent_id": parent_id,
"child_type": child_type,
"child_id": child_id,
**({"relation_type": relation_type} if relation_type else {}),
},
source_type="container_relation",
source_id=rel.id,
@@ -102,8 +124,9 @@ async def get_children(
parent_type: str,
parent_id: int,
child_type: str | None = None,
relation_type: str | None = None,
) -> list[ContainerRelation]:
"""Query children of a container, optionally filtered by child_type."""
"""Query children of a container, optionally filtered by child_type or relation_type."""
stmt = select(ContainerRelation).where(
ContainerRelation.parent_type == parent_type,
ContainerRelation.parent_id == parent_id,
@@ -111,6 +134,8 @@ async def get_children(
)
if child_type is not None:
stmt = stmt.where(ContainerRelation.child_type == child_type)
if relation_type is not None:
stmt = stmt.where(ContainerRelation.relation_type == relation_type)
stmt = stmt.order_by(
ContainerRelation.sort_order.asc(), ContainerRelation.id.asc()
@@ -119,23 +144,49 @@ async def get_children(
return list(result.scalars().all())
async def get_parents(
session: AsyncSession,
child_type: str,
child_id: int,
parent_type: str | None = None,
relation_type: str | None = None,
) -> list[ContainerRelation]:
"""Query parents of an entity, optionally filtered by parent_type or relation_type."""
stmt = select(ContainerRelation).where(
ContainerRelation.child_type == child_type,
ContainerRelation.child_id == child_id,
ContainerRelation.deleted_at.is_(None),
)
if parent_type is not None:
stmt = stmt.where(ContainerRelation.parent_type == parent_type)
if relation_type is not None:
stmt = stmt.where(ContainerRelation.relation_type == relation_type)
stmt = stmt.order_by(ContainerRelation.id.asc())
result = await session.execute(stmt)
return list(result.scalars().all())
async def detach_child(
session: AsyncSession,
parent_type: str,
parent_id: int,
child_type: str,
child_id: int,
relation_type: str | None = None,
) -> bool:
"""Soft-delete a ContainerRelation and emit container.child_detached event."""
result = await session.execute(
select(ContainerRelation).where(
ContainerRelation.parent_type == parent_type,
ContainerRelation.parent_id == parent_id,
ContainerRelation.child_type == child_type,
ContainerRelation.child_id == child_id,
ContainerRelation.deleted_at.is_(None),
)
stmt = select(ContainerRelation).where(
ContainerRelation.parent_type == parent_type,
ContainerRelation.parent_id == parent_id,
ContainerRelation.child_type == child_type,
ContainerRelation.child_id == child_id,
ContainerRelation.deleted_at.is_(None),
)
if relation_type is not None:
stmt = stmt.where(ContainerRelation.relation_type == relation_type)
result = await session.execute(stmt)
rel = result.scalar_one_or_none()
if not rel:
return False
@@ -153,6 +204,7 @@ async def detach_child(
"parent_id": parent_id,
"child_type": child_type,
"child_id": child_id,
**({"relation_type": rel.relation_type} if rel.relation_type else {}),
},
source_type="container_relation",
source_id=rel.id,

49
shared/sexp/components.py
View File

@@ -36,6 +36,10 @@ def load_shared_components() -> None:
register_components(_SEARCH_DESKTOP)
register_components(_MOBILE_FILTER)
register_components(_ORDER_SUMMARY_CARD)
# Relation-driven components
register_components(_RELATION_NAV)
register_components(_RELATION_ATTACH)
register_components(_RELATION_DETACH)
# ---------------------------------------------------------------------------
@@ -768,3 +772,48 @@ _ORDER_SUMMARY_CARD = r'''
(str (or currency "GBP") " " total-amount)
"\u2013"))))
'''
# ---------------------------------------------------------------------------
# ~relation-nav
# ---------------------------------------------------------------------------
_RELATION_NAV = '''
(defcomp ~relation-nav (&key href name icon nav-class relation-type)
(a :href href :class (or nav-class "flex items-center gap-3 rounded-lg py-2 px-3 text-sm text-stone-700 hover:bg-stone-100 transition-colors")
(when icon
(div :class "w-8 h-8 rounded bg-stone-200 flex items-center justify-center flex-shrink-0"
(i :class icon :aria-hidden "true")))
(div :class "flex-1 min-w-0"
(div :class "font-medium truncate" name))))
'''
# ---------------------------------------------------------------------------
# ~relation-attach
# ---------------------------------------------------------------------------
_RELATION_ATTACH = '''
(defcomp ~relation-attach (&key create-url label icon)
(a :href create-url
:hx-get create-url
:hx-target "#main-panel"
:hx-swap "outerHTML"
:hx-push-url "true"
:class "flex items-center gap-2 text-sm p-2 rounded hover:bg-stone-100 text-stone-500 hover:text-stone-700 transition-colors"
(when icon (i :class icon))
(span (or label "Add"))))
'''
# ---------------------------------------------------------------------------
# ~relation-detach
# ---------------------------------------------------------------------------
_RELATION_DETACH = '''
(defcomp ~relation-detach (&key detach-url name)
(button :hx-delete detach-url
:hx-confirm (str "Remove " (or name "this item") "?")
:class "text-red-500 hover:text-red-700 text-sm p-1 rounded hover:bg-red-50 transition-colors"
(i :class "fa fa-times" :aria-hidden "true")))
'''

73
shared/sexp/evaluator.py
View File

@@ -13,6 +13,7 @@ Special forms:
(lambda (params...) body) or (fn (params...) body)
(define name value)
(defcomp ~name (&key param...) body)
(defrelation :name :from "type" :to "type" :cardinality :card ...)
(begin expr...)
(quote expr)
(do expr...) — alias for begin
@@ -32,7 +33,7 @@ from __future__ import annotations
from typing import Any
from .types import Component, Keyword, Lambda, NIL, Symbol
from .types import Component, Keyword, Lambda, NIL, RelationDef, Symbol
from .primitives import _PRIMITIVES
@@ -429,6 +430,75 @@ def _sf_set_bang(expr: list, env: dict) -> Any:
return value
_VALID_CARDINALITIES = {"one-to-one", "one-to-many", "many-to-many"}
_VALID_NAV = {"submenu", "tab", "badge", "inline", "hidden"}
def _sf_defrelation(expr: list, env: dict) -> RelationDef:
"""``(defrelation :name :from "t" :to "t" :cardinality :card ...)``"""
if len(expr) < 2:
raise EvalError("defrelation requires a name")
name_kw = expr[1]
if not isinstance(name_kw, Keyword):
raise EvalError(f"defrelation name must be a keyword, got {type(name_kw).__name__}")
rel_name = name_kw.name
# Parse keyword args from remaining elements
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]
if isinstance(val, Keyword):
kwargs[key.name] = val.name
else:
kwargs[key.name] = _eval(val, env) if not isinstance(val, str) 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 EvalError(f"defrelation {rel_name} missing required :{field}")
card = kwargs["cardinality"]
if card not in _VALID_CARDINALITIES:
raise EvalError(
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 EvalError(
f"defrelation {rel_name}: invalid nav {nav!r}, "
f"expected one of {_VALID_NAV}"
)
defn = 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"),
)
from .relations import register_relation
register_relation(defn)
env[f"relation:{rel_name}"] = defn
return defn
_SPECIAL_FORMS: dict[str, Any] = {
"if": _sf_if,
"when": _sf_when,
@@ -442,6 +512,7 @@ _SPECIAL_FORMS: dict[str, Any] = {
"fn": _sf_lambda,
"define": _sf_define,
"defcomp": _sf_defcomp,
"defrelation": _sf_defrelation,
"begin": _sf_begin,
"do": _sf_begin,
"quote": _sf_quote,

2
shared/sexp/parser.py
View File

@@ -46,7 +46,7 @@ class Tokenizer:
COMMENT = re.compile(r";[^\n]*")
STRING = re.compile(r'"(?:[^"\\]|\\.)*"')
NUMBER = re.compile(r"-?(?:\d+\.?\d*|\.\d+)(?:[eE][+-]?\d+)?")
KEYWORD = re.compile(r":[a-zA-Z_][a-zA-Z0-9_-]*")
KEYWORD = re.compile(r":[a-zA-Z_][a-zA-Z0-9_>-]*")
# Symbols may start with alpha, _, or common operator chars, plus ~ for components,
# <> for the fragment symbol, and & for &key/&rest.
SYMBOL = re.compile(r"[a-zA-Z_~*+\-><=/!?&][a-zA-Z0-9_~*+\-><=/!?.:&]*")

101
shared/sexp/relations.py Normal file
View File

@@ -0,0 +1,101 @@
"""
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()``.
"""
from __future__ import annotations
from shared.sexp.types import RelationDef
# ---------------------------------------------------------------------------
# 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()
# ---------------------------------------------------------------------------
# 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.sexp.evaluator import evaluate
from shared.sexp.parser import parse
tree = parse(_BUILTIN_RELATIONS)
evaluate(tree)

74
shared/sexp/tests/test_components.py
View File

@@ -264,6 +264,80 @@ class TestErrorPage:
assert "<img" not in html
# ---------------------------------------------------------------------------
# ~relation-nav
# ---------------------------------------------------------------------------
class TestRelationNav:
def test_renders_link(self):
html = sexp(
'(~relation-nav :href "/market/farm/" :name "Farm Shop" :icon "fa fa-shopping-bag")',
)
assert 'href="/market/farm/"' in html
assert "Farm Shop" in html
assert "fa fa-shopping-bag" in html
def test_no_icon(self):
html = sexp(
'(~relation-nav :href "/cal/" :name "Events")',
)
assert 'href="/cal/"' in html
assert "Events" in html
assert "fa " not in html
def test_custom_nav_class(self):
html = sexp(
'(~relation-nav :href "/" :name "X" :nav-class "custom-class")',
**{"nav-class": "custom-class"},
)
assert 'class="custom-class"' in html
# ---------------------------------------------------------------------------
# ~relation-attach
# ---------------------------------------------------------------------------
class TestRelationAttach:
def test_renders_button(self):
html = sexp(
'(~relation-attach :create-url "/market/create/" :label "Add Market" :icon "fa fa-plus")',
**{"create-url": "/market/create/"},
)
assert 'href="/market/create/"' in html
assert 'hx-get="/market/create/"' in html
assert "Add Market" in html
assert "fa fa-plus" in html
def test_default_label(self):
html = sexp(
'(~relation-attach :create-url "/create/")',
**{"create-url": "/create/"},
)
assert "Add" in html
# ---------------------------------------------------------------------------
# ~relation-detach
# ---------------------------------------------------------------------------
class TestRelationDetach:
def test_renders_button(self):
html = sexp(
'(~relation-detach :detach-url "/api/unrelate" :name "Farm Shop")',
**{"detach-url": "/api/unrelate"},
)
assert 'hx-delete="/api/unrelate"' in html
assert 'hx-confirm="Remove Farm Shop?"' in html
assert "fa fa-times" in html
def test_default_name(self):
html = sexp(
'(~relation-detach :detach-url "/api/unrelate")',
**{"detach-url": "/api/unrelate"},
)
assert "this item" in html
# ---------------------------------------------------------------------------
# render_page() helper
# ---------------------------------------------------------------------------

245
shared/sexp/tests/test_relations.py Normal file
View File

@@ -0,0 +1,245 @@
"""Tests for the relation registry (Phase A)."""
import pytest
from shared.sexp.evaluator import evaluate, EvalError
from shared.sexp.parser import parse
from shared.sexp.relations import (
_RELATION_REGISTRY,
clear_registry,
get_relation,
load_relation_registry,
relations_from,
relations_to,
all_relations,
)
from shared.sexp.types import RelationDef
@pytest.fixture(autouse=True)
def _clean_registry():
"""Clear registry before each test."""
clear_registry()
# ---------------------------------------------------------------------------
# defrelation parsing
# ---------------------------------------------------------------------------
class TestDefrelation:
def test_basic_defrelation(self):
tree = parse('''
(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")
''')
result = evaluate(tree)
assert isinstance(result, RelationDef)
assert result.name == "page->market"
assert result.from_type == "page"
assert result.to_type == "market"
assert result.cardinality == "one-to-many"
assert result.inverse == "market->page"
assert result.nav == "submenu"
assert result.nav_icon == "fa fa-shopping-bag"
assert result.nav_label == "markets"
def test_defrelation_registered(self):
tree = parse('''
(defrelation :a->b
:from "a" :to "b" :cardinality :one-to-one :nav :hidden)
''')
evaluate(tree)
assert get_relation("a->b") is not None
assert get_relation("a->b").cardinality == "one-to-one"
def test_defrelation_one_to_one(self):
tree = parse('''
(defrelation :page->menu_node
:from "page" :to "menu_node"
:cardinality :one-to-one :nav :hidden)
''')
result = evaluate(tree)
assert result.cardinality == "one-to-one"
assert result.inverse is None
assert result.nav == "hidden"
def test_defrelation_many_to_many(self):
tree = parse('''
(defrelation :post->entry
:from "post" :to "calendar_entry"
:cardinality :many-to-many
:inverse :entry->post
:nav :inline
:nav-icon "fa fa-file-alt"
:nav-label "events")
''')
result = evaluate(tree)
assert result.cardinality == "many-to-many"
def test_default_nav_is_hidden(self):
tree = parse('''
(defrelation :x->y
:from "x" :to "y" :cardinality :one-to-many)
''')
result = evaluate(tree)
assert result.nav == "hidden"
def test_invalid_cardinality_raises(self):
tree = parse('''
(defrelation :bad
:from "a" :to "b" :cardinality :wrong)
''')
with pytest.raises(EvalError, match="invalid cardinality"):
evaluate(tree)
def test_invalid_nav_raises(self):
tree = parse('''
(defrelation :bad
:from "a" :to "b" :cardinality :one-to-one :nav :bogus)
''')
with pytest.raises(EvalError, match="invalid nav"):
evaluate(tree)
def test_missing_from_raises(self):
tree = parse('''
(defrelation :bad :to "b" :cardinality :one-to-one)
''')
with pytest.raises(EvalError, match="missing required :from"):
evaluate(tree)
def test_missing_to_raises(self):
tree = parse('''
(defrelation :bad :from "a" :cardinality :one-to-one)
''')
with pytest.raises(EvalError, match="missing required :to"):
evaluate(tree)
def test_missing_cardinality_raises(self):
tree = parse('''
(defrelation :bad :from "a" :to "b")
''')
with pytest.raises(EvalError, match="missing required :cardinality"):
evaluate(tree)
def test_name_must_be_keyword(self):
tree = parse('(defrelation "not-keyword" :from "a" :to "b" :cardinality :one-to-one)')
with pytest.raises(EvalError, match="must be a keyword"):
evaluate(tree)
# ---------------------------------------------------------------------------
# Registry queries
# ---------------------------------------------------------------------------
class TestRegistry:
def _load_sample(self):
tree = parse('''
(begin
(defrelation :page->market
:from "page" :to "market" :cardinality :one-to-many
:nav :submenu :nav-icon "fa fa-shopping-bag" :nav-label "markets")
(defrelation :page->calendar
:from "page" :to "calendar" :cardinality :one-to-many
:nav :submenu :nav-icon "fa fa-calendar" :nav-label "calendars")
(defrelation :post->entry
:from "post" :to "calendar_entry" :cardinality :many-to-many
:nav :inline)
(defrelation :page->menu_node
:from "page" :to "menu_node" :cardinality :one-to-one
:nav :hidden))
''')
evaluate(tree)
def test_get_relation(self):
self._load_sample()
rel = get_relation("page->market")
assert rel is not None
assert rel.to_type == "market"
def test_get_relation_not_found(self):
assert get_relation("nonexistent") is None
def test_relations_from_page(self):
self._load_sample()
rels = relations_from("page")
names = {r.name for r in rels}
assert names == {"page->market", "page->calendar", "page->menu_node"}
def test_relations_from_post(self):
self._load_sample()
rels = relations_from("post")
assert len(rels) == 1
assert rels[0].name == "post->entry"
def test_relations_from_empty(self):
self._load_sample()
assert relations_from("nonexistent") == []
def test_relations_to_market(self):
self._load_sample()
rels = relations_to("market")
assert len(rels) == 1
assert rels[0].name == "page->market"
def test_relations_to_calendar_entry(self):
self._load_sample()
rels = relations_to("calendar_entry")
assert len(rels) == 1
assert rels[0].name == "post->entry"
def test_all_relations(self):
self._load_sample()
assert len(all_relations()) == 4
def test_clear_registry(self):
self._load_sample()
assert len(all_relations()) == 4
clear_registry()
assert len(all_relations()) == 0
# ---------------------------------------------------------------------------
# load_relation_registry() — built-in definitions
# ---------------------------------------------------------------------------
class TestLoadBuiltins:
def test_loads_builtin_relations(self):
load_relation_registry()
assert get_relation("page->market") is not None
assert get_relation("page->calendar") is not None
assert get_relation("post->calendar_entry") is not None
assert get_relation("page->menu_node") is not None
def test_builtin_page_market(self):
load_relation_registry()
rel = get_relation("page->market")
assert rel.from_type == "page"
assert rel.to_type == "market"
assert rel.cardinality == "one-to-many"
assert rel.inverse == "market->page"
assert rel.nav == "submenu"
assert rel.nav_icon == "fa fa-shopping-bag"
def test_builtin_post_entry(self):
load_relation_registry()
rel = get_relation("post->calendar_entry")
assert rel.cardinality == "many-to-many"
assert rel.nav == "inline"
def test_builtin_page_menu_node(self):
load_relation_registry()
rel = get_relation("page->menu_node")
assert rel.cardinality == "one-to-one"
assert rel.nav == "hidden"
def test_frozen_dataclass(self):
load_relation_registry()
rel = get_relation("page->market")
with pytest.raises(AttributeError):
rel.name = "changed"

22
shared/sexp/types.py
View File

@@ -148,9 +148,29 @@ class Component:
return f"<Component ~{self.name}({', '.join(self.params)})>"
# ---------------------------------------------------------------------------
# RelationDef
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class RelationDef:
"""A declared relation between two entity types.
Created by ``(defrelation :name ...)`` s-expressions.
"""
name: str # "page->market"
from_type: str # "page"
to_type: str # "market"
cardinality: str # "one-to-one" | "one-to-many" | "many-to-many"
inverse: str | None # "market->page"
nav: str # "submenu" | "tab" | "badge" | "inline" | "hidden"
nav_icon: str | None # "fa fa-shopping-bag"
nav_label: str | None # "markets"
# ---------------------------------------------------------------------------
# Type alias
# ---------------------------------------------------------------------------
# An s-expression value after evaluation
SExp = int | float | str | bool | Symbol | Keyword | Lambda | Component | list | dict | _Nil | None
SExp = int | float | str | bool | Symbol | Keyword | Lambda | Component | RelationDef | list | dict | _Nil | None