giles
278ae3e8f6
SxExpr is now a str subclass so it works everywhere a plain string does (join, isinstance, f-strings) while serialize() still emits it unquoted. sx_call() and all internal render functions (_render_to_sx, async_eval_to_sx, etc.) return SxExpr, eliminating the "forgot to wrap" bug class that caused the sx_content leak and list serialization bugs. - Phase 0: SxExpr(str) with .source property, __add__/__radd__ - Phase 1: sx_call returns SxExpr (drop-in, all 200+ sites unchanged) - Phase 2: async_eval_to_sx, async_eval_slot_to_sx, _render_to_sx, mobile_menu_sx return SxExpr; remove isinstance(str) workaround - Phase 3: Remove ~150 redundant SxExpr() wrappings across 45 files - Phase 4: serialize() docstring, handler return docs, ;; returns: sx Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
250 lines
8.9 KiB
Python
250 lines
8.9 KiB
Python
"""
|
|
Declarative handler registry and blueprint factory.
|
|
|
|
Supports ``defhandler`` s-expressions that define fragment handlers
|
|
in .sx files instead of Python. Each handler is a self-contained
|
|
s-expression with a bounded primitive vocabulary, providing a clear
|
|
security boundary and AI legibility.
|
|
|
|
Usage::
|
|
|
|
from shared.sx.handlers import create_handler_blueprint, load_handler_file
|
|
|
|
# Load handler definitions from .sx files
|
|
load_handler_file("blog/sx/handlers/link-card.sx", "blog")
|
|
|
|
# Create a blueprint that dispatches to both sx and Python handlers
|
|
bp = create_handler_blueprint("blog")
|
|
bp.add_python_handler("nav-tree", nav_tree_handler_fn)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from typing import Any, Callable, Awaitable
|
|
|
|
from .types import HandlerDef
|
|
|
|
logger = logging.getLogger("sx.handlers")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry — service → handler-name → HandlerDef
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_HANDLER_REGISTRY: dict[str, dict[str, HandlerDef]] = {}
|
|
|
|
|
|
def register_handler(service: str, handler_def: HandlerDef) -> None:
|
|
"""Register a handler definition for a service."""
|
|
if service not in _HANDLER_REGISTRY:
|
|
_HANDLER_REGISTRY[service] = {}
|
|
_HANDLER_REGISTRY[service][handler_def.name] = handler_def
|
|
logger.debug("Registered handler %s:%s", service, handler_def.name)
|
|
|
|
|
|
def get_handler(service: str, name: str) -> HandlerDef | None:
|
|
"""Look up a registered handler by service and name."""
|
|
return _HANDLER_REGISTRY.get(service, {}).get(name)
|
|
|
|
|
|
def get_all_handlers(service: str) -> dict[str, HandlerDef]:
|
|
"""Return all handlers for a service."""
|
|
return dict(_HANDLER_REGISTRY.get(service, {}))
|
|
|
|
|
|
def clear_handlers(service: str | None = None) -> None:
|
|
"""Clear handler registry. If service given, clear only that service."""
|
|
if service is None:
|
|
_HANDLER_REGISTRY.clear()
|
|
else:
|
|
_HANDLER_REGISTRY.pop(service, None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Loading — parse .sx files and collect HandlerDef instances
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def load_handler_file(filepath: str, service_name: str) -> list[HandlerDef]:
|
|
"""Parse an .sx file, evaluate it, and register any HandlerDef values."""
|
|
from .parser import parse_all
|
|
from .evaluator import _eval as _raw_eval, _trampoline
|
|
_eval = lambda expr, env: _trampoline(_raw_eval(expr, env))
|
|
from .jinja_bridge import get_component_env
|
|
|
|
with open(filepath, encoding="utf-8") as f:
|
|
source = f.read()
|
|
|
|
# Seed env with component definitions so handlers can reference components
|
|
env = dict(get_component_env())
|
|
exprs = parse_all(source)
|
|
handlers: list[HandlerDef] = []
|
|
|
|
for expr in exprs:
|
|
_eval(expr, env)
|
|
|
|
# Collect all HandlerDef values from the env
|
|
for key, val in env.items():
|
|
if isinstance(val, HandlerDef):
|
|
register_handler(service_name, val)
|
|
handlers.append(val)
|
|
|
|
return handlers
|
|
|
|
|
|
def load_handler_dir(directory: str, service_name: str) -> list[HandlerDef]:
|
|
"""Load all .sx files from a directory and register handlers."""
|
|
import glob as glob_mod
|
|
handlers: list[HandlerDef] = []
|
|
for filepath in sorted(glob_mod.glob(os.path.join(directory, "*.sx"))):
|
|
handlers.extend(load_handler_file(filepath, service_name))
|
|
return handlers
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Handler execution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def execute_handler(
|
|
handler_def: HandlerDef,
|
|
service_name: str,
|
|
args: dict[str, str] | None = None,
|
|
) -> str:
|
|
"""Execute a declarative handler and return SX wire format (``SxExpr``).
|
|
|
|
Uses the async evaluator so I/O primitives (``query``, ``service``,
|
|
``request-arg``, etc.) are awaited inline within control flow.
|
|
|
|
Returns ``SxExpr`` — pre-built sx source. Callers like
|
|
``fetch_fragment`` check ``content-type: text/sx`` and wrap the
|
|
response in ``SxExpr`` when consuming cross-service fragments.
|
|
|
|
1. Build env from component env + handler closure
|
|
2. Bind handler params from args (typically request.args)
|
|
3. Evaluate via ``async_eval_to_sx`` (I/O inline, components serialized)
|
|
4. Return ``SxExpr`` wire format
|
|
"""
|
|
from .jinja_bridge import get_component_env, _get_request_context
|
|
from .async_eval import async_eval_to_sx
|
|
from .types import NIL
|
|
|
|
if args is None:
|
|
args = {}
|
|
|
|
# Build environment
|
|
env = dict(get_component_env())
|
|
env.update(handler_def.closure)
|
|
|
|
# Bind handler params from request args
|
|
for param in handler_def.params:
|
|
env[param] = args.get(param, args.get(param.replace("-", "_"), NIL))
|
|
|
|
# Get request context for I/O primitives
|
|
ctx = _get_request_context()
|
|
|
|
# Async eval → sx source — I/O primitives are awaited inline,
|
|
# but component/tag calls serialize to sx wire format (not HTML).
|
|
return await async_eval_to_sx(handler_def.body, env, ctx)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Blueprint factory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def create_handler_blueprint(service_name: str) -> Any:
|
|
"""Create a Quart Blueprint that dispatches fragment requests to
|
|
both sx-defined handlers and Python handler functions.
|
|
|
|
Usage::
|
|
|
|
bp = create_handler_blueprint("blog")
|
|
bp.add_python_handler("nav-tree", my_python_handler)
|
|
app.register_blueprint(bp)
|
|
"""
|
|
from quart import Blueprint, Response, request
|
|
from shared.infrastructure.fragments import FRAGMENT_HEADER
|
|
|
|
bp = Blueprint(
|
|
f"sx_handlers_{service_name}",
|
|
__name__,
|
|
url_prefix="/internal/fragments",
|
|
)
|
|
|
|
# Python-side handler overrides
|
|
_python_handlers: dict[str, Callable[[], Awaitable[str]]] = {}
|
|
|
|
@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):
|
|
# 1. Check Python handlers first (manual overrides)
|
|
py_handler = _python_handlers.get(fragment_type)
|
|
if py_handler is not None:
|
|
result = await py_handler()
|
|
return Response(result, status=200, content_type="text/sx")
|
|
|
|
# 2. Check sx handler registry
|
|
handler_def = get_handler(service_name, fragment_type)
|
|
if handler_def is not None:
|
|
result = await execute_handler(
|
|
handler_def,
|
|
service_name,
|
|
args=dict(request.args),
|
|
)
|
|
return Response(result, status=200, content_type="text/sx")
|
|
|
|
# 3. No handler found — return empty
|
|
return Response("", status=200, content_type="text/sx")
|
|
|
|
def add_python_handler(name: str, fn: Callable[[], Awaitable[str]]) -> None:
|
|
"""Register a Python async function as a fragment handler."""
|
|
_python_handlers[name] = fn
|
|
|
|
bp.add_python_handler = add_python_handler # type: ignore[attr-defined]
|
|
bp._python_handlers = _python_handlers # type: ignore[attr-defined]
|
|
|
|
return bp
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Direct app mount — replaces per-service fragment blueprint boilerplate
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def auto_mount_fragment_handlers(app: Any, service_name: str) -> Callable:
|
|
"""Mount ``/internal/fragments/<type>`` directly on the app.
|
|
|
|
Returns an ``add_handler(name, fn, content_type)`` function for
|
|
registering Python handler overrides (checked before SX handlers).
|
|
"""
|
|
from quart import Response, request
|
|
from shared.infrastructure.fragments import FRAGMENT_HEADER
|
|
|
|
python_handlers: dict[str, Callable[[], Awaitable[str]]] = {}
|
|
html_types: set[str] = set()
|
|
|
|
@app.get("/internal/fragments/<fragment_type>")
|
|
async def _fragment_dispatch(fragment_type: str):
|
|
if not request.headers.get(FRAGMENT_HEADER):
|
|
return Response("", status=403)
|
|
py = python_handlers.get(fragment_type)
|
|
if py is not None:
|
|
result = await py()
|
|
ct = "text/html" if fragment_type in html_types else "text/sx"
|
|
return Response(result, status=200, content_type=ct)
|
|
hdef = get_handler(service_name, fragment_type)
|
|
if hdef is not None:
|
|
result = await execute_handler(hdef, service_name, args=dict(request.args))
|
|
return Response(result, status=200, content_type="text/sx")
|
|
return Response("", status=200, content_type="text/sx")
|
|
|
|
def add_handler(name: str, fn: Callable[[], Awaitable[str]], content_type: str = "text/sx") -> None:
|
|
python_handlers[name] = fn
|
|
if content_type == "text/html":
|
|
html_types.add(name)
|
|
|
|
return add_handler
|