Enforce SX boundary contract via boundary.sx spec + runtime validation
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 2m33s

Add boundary.sx declaring all 34 I/O primitives, 32 page helpers, and 9
allowed boundary types. Runtime validation in boundary.py checks every
registration against the spec — undeclared primitives/helpers crash at
startup with SX_BOUNDARY_STRICT=1 (now set in both dev and prod).

Key changes:
- Move 5 I/O-in-disguise primitives (app-url, asset-url, config,
  jinja-global, relations-from) from primitives.py to primitives_io.py
- Remove duplicate url-for/route-prefix from primitives.py (already in IO)
- Fix parse-datetime to return ISO string instead of raw datetime
- Add datetime→isoformat conversion in _convert_result at the edge
- Wrap page helper return values with boundary type validation
- Replace all SxExpr(f"...") patterns with sx_call() or _sx_fragment()
- Add assert declaration to primitives.sx

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-05 23:50:02 +00:00
parent 54adc9c216
commit 04366990ec
21 changed files with 1342 additions and 415 deletions

View File

@@ -59,6 +59,11 @@ IO_PRIMITIVES: frozenset[str] = frozenset({
"events-slot-ctx",
"events-ticket-type-ctx",
"market-header-ctx",
"app-url",
"asset-url",
"config",
"jinja-global",
"relations-from",
})
@@ -258,10 +263,19 @@ def _dto_to_dict(obj: Any) -> dict[str, Any]:
def _convert_result(result: Any) -> Any:
"""Convert a service method result for sx consumption."""
"""Convert a service method result for sx consumption.
Converts DTOs/dataclasses to dicts, datetimes to ISO strings,
and ensures only SX-typed values cross the boundary.
"""
if result is None:
from .types import NIL
return NIL
if isinstance(result, (int, float, str, bool)):
return result
# datetime → ISO string at the edge
if hasattr(result, "isoformat") and callable(result.isoformat):
return result.isoformat()
if isinstance(result, dict):
return {k: _convert_result(v) for k, v in result.items()}
if isinstance(result, tuple):
@@ -273,7 +287,7 @@ def _convert_result(result: Any) -> Any:
return [
_dto_to_dict(item)
if hasattr(item, "__dataclass_fields__") or hasattr(item, "_asdict")
else item
else _convert_result(item)
for item in result
]
return result
@@ -474,14 +488,14 @@ async def _io_account_nav_ctx(
from quart import g
from .types import NIL
from .parser import SxExpr
from .helpers import sx_call
val = getattr(g, "account_nav", None)
if not val:
return NIL
if isinstance(val, SxExpr):
return val
# HTML string → wrap for SX rendering
escaped = str(val).replace("\\", "\\\\").replace('"', '\\"')
return SxExpr(f'(~rich-text :html "{escaped}")')
return sx_call("rich-text", html=str(val))
async def _io_app_rights(
@@ -873,6 +887,67 @@ async def _io_events_ticket_type_ctx(
}
async def _io_app_url(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> str:
"""``(app-url "blog" "/my-post/")`` → full URL for service."""
if not args:
raise ValueError("app-url requires a service name")
from shared.infrastructure.urls import app_url
service = str(args[0])
path = str(args[1]) if len(args) > 1 else "/"
return app_url(service, path)
async def _io_asset_url(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> str:
"""``(asset-url "/img/logo.png")`` → versioned static URL."""
from shared.infrastructure.urls import asset_url
path = str(args[0]) if args else ""
return asset_url(path)
async def _io_config(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> Any:
"""``(config "key")`` → shared.config.config()[key]."""
if not args:
raise ValueError("config requires a key")
from shared.config import config
cfg = config()
return cfg.get(str(args[0]))
async def _io_jinja_global(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> Any:
"""``(jinja-global "key")`` → current_app.jinja_env.globals[key]."""
if not args:
raise ValueError("jinja-global requires a key")
from quart import current_app
key = str(args[0])
default = args[1] if len(args) > 1 else None
return current_app.jinja_env.globals.get(key, default)
async def _io_relations_from(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> list[dict]:
"""``(relations-from "page")`` → list of RelationDef dicts."""
if not args:
raise ValueError("relations-from requires an entity type")
from shared.sx.relations import relations_from
return [
{
"name": d.name, "from_type": d.from_type, "to_type": d.to_type,
"cardinality": d.cardinality, "nav": d.nav,
"nav_icon": d.nav_icon, "nav_label": d.nav_label,
}
for d in relations_from(str(args[0]))
]
async def _io_market_header_ctx(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> dict[str, Any]:
@@ -966,4 +1041,20 @@ _IO_HANDLERS: dict[str, Any] = {
"events-slot-ctx": _io_events_slot_ctx,
"events-ticket-type-ctx": _io_events_ticket_type_ctx,
"market-header-ctx": _io_market_header_ctx,
"app-url": _io_app_url,
"asset-url": _io_asset_url,
"config": _io_config,
"jinja-global": _io_jinja_global,
"relations-from": _io_relations_from,
}
# Validate all I/O handlers are declared in boundary.sx
def _validate_io_handlers() -> None:
from .boundary import validate_io
for name in _IO_HANDLERS:
validate_io(name)
for name in IO_PRIMITIVES:
if name not in _IO_HANDLERS:
validate_io(name)
_validate_io_handlers()