Migrate remaining 7 ref endpoints to SX, add :returns type annotations

Add 14 new IO primitives to boundary.sx: web interop (request-form,
request-json, request-header, request-content-type, request-args-all,
request-form-all, request-headers-all, request-file-name), response
manipulation (set-response-header, set-response-status), ephemeral
state (state-get, state-set!), and timing (now, sleep).

All 19 reference handlers now have :returns type annotations using
types.sx vocabulary. Response meta (headers/status) flows through
context vars, applied by register_route_handlers after execution.

Only SSE endpoint remains in Python (async generator paradigm).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-12 00:14:40 +00:00
parent fba84540e2
commit 575d100f67
9 changed files with 338 additions and 103 deletions

View File

@@ -713,84 +713,9 @@ def register(url_prefix: str = "/") -> Blueprint:
# Reference API endpoints — remaining Python-only
#
# Most reference endpoints migrated to sx/sx/handlers/ref-api.sx.
# These remain because they need Python-specific features:
# - File upload access (request.files)
# - Dynamic all-params iteration
# - Stateful counters with non-200 responses
# - SSE streaming
# - Custom response headers
# SSE stays in Python — fundamentally different paradigm (async generator).
# ------------------------------------------------------------------
def _ref_wire(wire_id: str, sx_src: str) -> str:
"""Build OOB swap showing the wire response text."""
from sxc.pages.renders import _oob_code
return _oob_code(f"ref-wire-{wire_id}", sx_src)
@csrf_exempt
@bp.post("/geography/hypermedia/reference/api/upload-name")
async def ref_upload_name():
from shared.sx.helpers import sx_response
files = await request.files
f = files.get("file")
name = f.filename if f else "(no file)"
sx_src = f'(span :class "text-stone-800 text-sm" "Received: " (strong "{name}"))'
oob = _ref_wire("sx-encoding", sx_src)
return sx_response(f'(<> {sx_src} {oob})')
@bp.get("/geography/hypermedia/reference/api/echo-headers")
async def ref_echo_headers():
from shared.sx.helpers import sx_response
custom = [(k, v) for k, v in request.headers if k.lower().startswith("x-")]
if not custom:
sx_src = '(span :class "text-stone-400 text-sm" "No custom headers received.")'
else:
items = " ".join(
f'(li (strong "{k}") ": " "{v}")' for k, v in custom)
sx_src = f'(ul :class "text-sm text-stone-700 space-y-1" {items})'
oob = _ref_wire("sx-headers", sx_src)
return sx_response(f'(<> {sx_src} {oob})')
@bp.get("/geography/hypermedia/reference/api/echo-vals")
async def ref_echo_vals_get():
from shared.sx.helpers import sx_response
vals = list(request.args.items())
if not vals:
sx_src = '(span :class "text-stone-400 text-sm" "No values received.")'
else:
items = " ".join(
f'(li (strong "{k}") ": " "{v}")' for k, v in vals)
sx_src = f'(ul :class "text-sm text-stone-700 space-y-1" {items})'
oob_include = _ref_wire("sx-include", sx_src)
return sx_response(f'(<> {sx_src} {oob_include})')
@csrf_exempt
@bp.post("/geography/hypermedia/reference/api/echo-vals")
async def ref_echo_vals_post():
from shared.sx.helpers import sx_response
form = await request.form
vals = list(form.items())
if not vals:
sx_src = '(span :class "text-stone-400 text-sm" "No values received.")'
else:
items = " ".join(
f'(li (strong "{k}") ": " "{v}")' for k, v in vals)
sx_src = f'(ul :class "text-sm text-stone-700 space-y-1" {items})'
oob = _ref_wire("sx-vals", sx_src)
return sx_response(f'(<> {sx_src} {oob})')
_ref_flaky = {"n": 0}
@bp.get("/geography/hypermedia/reference/api/flaky")
async def ref_flaky():
from shared.sx.helpers import sx_response
_ref_flaky["n"] += 1
n = _ref_flaky["n"]
if n % 3 != 0:
return Response("", status=503, content_type="text/plain")
sx_src = f'(span :class "text-emerald-700 text-sm" "Success on attempt " "{n}" "!")'
oob = _ref_wire("sx-retry", sx_src)
return sx_response(f'(<> {sx_src} {oob})')
@bp.get("/geography/hypermedia/reference/api/sse-time")
async def ref_sse_time():
async def generate():
@@ -930,24 +855,4 @@ def register(url_prefix: str = "/") -> Blueprint:
)
return sx_response(sx_src)
# --- Header demos ---
@bp.get("/geography/hypermedia/reference/api/trigger-event")
async def ref_trigger_event():
from shared.sx.helpers import sx_response
now = datetime.now().strftime("%H:%M:%S")
sx_src = f'(span :class "text-stone-800 text-sm" "Loaded at " (strong "{now}") " — check the border!")'
resp = sx_response(sx_src)
resp.headers["SX-Trigger"] = "showNotice"
return resp
@bp.get("/geography/hypermedia/reference/api/retarget")
async def ref_retarget():
from shared.sx.helpers import sx_response
now = datetime.now().strftime("%H:%M:%S")
sx_src = f'(span :class "text-violet-700 text-sm" "Retargeted at " (strong "{now}"))'
resp = sx_response(sx_src)
resp.headers["SX-Retarget"] = "#ref-hdr-retarget-alt"
return resp
return bp