Add macros, declarative handlers (defhandler), and convert all fragment routes to sx
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled
Phase 1 — Macros: defmacro + quasiquote syntax (`, ,, ,@) in parser,
evaluator, HTML renderer, and JS mirror. Macro type, expansion, and
round-trip serialization.
Phase 2 — Expanded primitives: app-url, url-for, asset-url, config,
format-date, parse-int (pure); service, request-arg, request-path,
nav-tree, get-children (I/O); jinja-global, relations-from (pure).
Updated _io_service to accept (service "registry-name" "method" :kwargs)
with auto kebab→snake conversion. DTO-to-dict now expands datetime fields
into year/month/day convenience keys. Tuple returns converted to lists.
Phase 3 — Declarative handlers: HandlerDef type, defhandler special form,
handler registry (service → name → HandlerDef), async evaluator+renderer
(async_eval.py) that awaits I/O primitives inline within control flow.
Handler loading from .sx files, execute_handler, blueprint factory.
Phase 4 — Convert all fragment routes: 13 Python fragment handlers across
8 services replaced with declarative .sx handler files. All routes.py
simplified to uniform sx dispatch pattern. Two Jinja HTML handlers
(events/container-cards, events/account-page) kept as Python.
New files: shared/sx/async_eval.py, shared/sx/handlers.py,
shared/sx/tests/test_handlers.py, plus 13 handler .sx files under
{service}/sx/handlers/. MarketService.product_by_slug() added.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
205
shared/sx/handlers.py
Normal file
205
shared/sx/handlers.py
Normal file
@@ -0,0 +1,205 @@
|
||||
"""
|
||||
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
|
||||
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 rendered sx/HTML string.
|
||||
|
||||
Uses the async evaluator+renderer so I/O primitives (``query``,
|
||||
``service``, ``request-arg``, etc.) are awaited inline within
|
||||
control flow — no collect-then-substitute limitations.
|
||||
|
||||
1. Build env from component env + handler closure
|
||||
2. Bind handler params from args (typically request.args)
|
||||
3. Evaluate + render via async_render (handles I/O inline)
|
||||
4. Return rendered string
|
||||
"""
|
||||
from .jinja_bridge import get_component_env, _get_request_context
|
||||
from .async_eval import async_render
|
||||
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+render — I/O primitives are awaited inline
|
||||
return await async_render(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
|
||||
Reference in New Issue
Block a user