Extend defhandler with :path/:method/:csrf, migrate 12 ref endpoints to SX

defhandler now supports keyword options for public route registration:
  (defhandler name :path "/..." :method :post :csrf false (&key) body)

Infrastructure: forms.sx parses options, HandlerDef stores path/method/csrf,
register_route_handlers() mounts path-based handlers as app routes.

New IO primitives (boundary.sx "Web interop" section): now, sleep,
request-form, request-json, request-header, request-content-type.

First migration: 12 reference API endpoints from Python f-string SX
to declarative .sx handlers in sx/sx/handlers/ref-api.sx.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-11 23:48:05 +00:00
parent 524c99e4ff
commit fba84540e2
10 changed files with 468 additions and 133 deletions

View File

@@ -297,6 +297,81 @@ async def _io_g(
return getattr(g, key, None)
@register_io_handler("now")
async def _io_now(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> str:
"""``(now)`` or ``(now "%H:%M:%S")`` → formatted timestamp string."""
from datetime import datetime
fmt = str(args[0]) if args else None
dt = datetime.now()
return dt.strftime(fmt) if fmt else dt.isoformat()
@register_io_handler("sleep")
async def _io_sleep(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> Any:
"""``(sleep 800)`` → pause for 800ms."""
import asyncio
from .types import NIL
if not args:
raise ValueError("sleep requires milliseconds")
ms = int(args[0])
await asyncio.sleep(ms / 1000.0)
return NIL
@register_io_handler("request-form")
async def _io_request_form(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> Any:
"""``(request-form "name" default?)`` → read a form field."""
if not args:
raise ValueError("request-form requires a field name")
from quart import request
from .types import NIL
name = str(args[0])
default = args[1] if len(args) > 1 else NIL
form = await request.form
return form.get(name, default)
@register_io_handler("request-json")
async def _io_request_json(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> Any:
"""``(request-json)`` → JSON body as dict, or nil."""
from quart import request
from .types import NIL
data = await request.get_json(silent=True)
return data if data is not None else NIL
@register_io_handler("request-header")
async def _io_request_header(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> Any:
"""``(request-header "name" default?)`` → request header value."""
if not args:
raise ValueError("request-header requires a header name")
from quart import request
from .types import NIL
name = str(args[0])
default = args[1] if len(args) > 1 else NIL
return request.headers.get(name, default)
@register_io_handler("request-content-type")
async def _io_request_content_type(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
) -> Any:
"""``(request-content-type)`` → content-type string or nil."""
from quart import request
from .types import NIL
return request.content_type or NIL
@register_io_handler("csrf-token")
async def _io_csrf_token(
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext