giles
a8c0741f54
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 4m5s
- Add sx_content to _post_to_edit_dict so edit page receives existing content - Add SX/Koenig editor tabs, sx-editor mount point, and SxEditor.mount init - Only pass sx_content to writer_update when form field is present (prevents accidental clearing when editing via Koenig-only path) - Add csrf_exempt to example API POST/DELETE/PUT demo endpoints - Add defpage infrastructure (pages.py, layouts.py) and sx docs page definitions - Add defhandler definitions for example API handlers (examples.sx) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
363 lines
12 KiB
Python
363 lines
12 KiB
Python
"""
|
|
Declarative page registry and blueprint mounting.
|
|
|
|
Supports ``defpage`` s-expressions that define GET route handlers
|
|
in .sx files instead of Python. Each page is a self-contained
|
|
declaration with path, auth, layout, data, and content slots.
|
|
|
|
Usage::
|
|
|
|
from shared.sx.pages import load_page_file, mount_pages
|
|
|
|
# Load page definitions from .sx files
|
|
load_page_file("blog/sx/pages/admin.sx", "blog")
|
|
|
|
# Mount page routes onto an existing blueprint
|
|
mount_pages(bp, "blog")
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from typing import Any
|
|
|
|
from .types import PageDef
|
|
|
|
logger = logging.getLogger("sx.pages")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry — service → page-name → PageDef
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_PAGE_REGISTRY: dict[str, dict[str, PageDef]] = {}
|
|
_PAGE_HELPERS: dict[str, dict[str, Any]] = {} # service → name → callable
|
|
|
|
|
|
def register_page(service: str, page_def: PageDef) -> None:
|
|
"""Register a page definition for a service."""
|
|
if service not in _PAGE_REGISTRY:
|
|
_PAGE_REGISTRY[service] = {}
|
|
_PAGE_REGISTRY[service][page_def.name] = page_def
|
|
logger.debug("Registered page %s:%s path=%s", service, page_def.name, page_def.path)
|
|
|
|
|
|
def get_page(service: str, name: str) -> PageDef | None:
|
|
"""Look up a registered page by service and name."""
|
|
return _PAGE_REGISTRY.get(service, {}).get(name)
|
|
|
|
|
|
def get_all_pages(service: str) -> dict[str, PageDef]:
|
|
"""Return all pages for a service."""
|
|
return dict(_PAGE_REGISTRY.get(service, {}))
|
|
|
|
|
|
def clear_pages(service: str | None = None) -> None:
|
|
"""Clear page registry. If service given, clear only that service."""
|
|
if service is None:
|
|
_PAGE_REGISTRY.clear()
|
|
else:
|
|
_PAGE_REGISTRY.pop(service, None)
|
|
|
|
|
|
def register_page_helpers(service: str, helpers: dict[str, Any]) -> None:
|
|
"""Register Python functions available in defpage content expressions.
|
|
|
|
These are injected into the evaluation environment when executing
|
|
defpage content, allowing defpage to call into Python::
|
|
|
|
register_page_helpers("sx", {
|
|
"docs-content": docs_content_sx,
|
|
"reference-content": reference_content_sx,
|
|
})
|
|
|
|
Then in .sx::
|
|
|
|
(defpage docs-page
|
|
:path "/docs/<slug>"
|
|
:auth :public
|
|
:content (docs-content slug))
|
|
"""
|
|
if service not in _PAGE_HELPERS:
|
|
_PAGE_HELPERS[service] = {}
|
|
_PAGE_HELPERS[service].update(helpers)
|
|
|
|
|
|
def get_page_helpers(service: str) -> dict[str, Any]:
|
|
"""Return registered page helpers for a service."""
|
|
return dict(_PAGE_HELPERS.get(service, {}))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Loading — parse .sx files and collect PageDef instances
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def load_page_file(filepath: str, service_name: str) -> list[PageDef]:
|
|
"""Parse an .sx file, evaluate it, and register any PageDef 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 pages can reference components
|
|
env = dict(get_component_env())
|
|
exprs = parse_all(source)
|
|
pages: list[PageDef] = []
|
|
|
|
for expr in exprs:
|
|
_eval(expr, env)
|
|
|
|
# Collect all PageDef values from the env
|
|
for key, val in env.items():
|
|
if isinstance(val, PageDef):
|
|
register_page(service_name, val)
|
|
pages.append(val)
|
|
|
|
return pages
|
|
|
|
|
|
def load_page_dir(directory: str, service_name: str) -> list[PageDef]:
|
|
"""Load all .sx files from a directory and register pages."""
|
|
import glob as glob_mod
|
|
pages: list[PageDef] = []
|
|
for filepath in sorted(glob_mod.glob(os.path.join(directory, "*.sx"))):
|
|
pages.extend(load_page_file(filepath, service_name))
|
|
return pages
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Page execution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _eval_slot(expr: Any, env: dict, ctx: Any,
|
|
async_eval_fn: Any, async_eval_to_sx_fn: Any) -> str:
|
|
"""Evaluate a page slot expression and return an sx source string.
|
|
|
|
If the expression evaluates to a plain string (e.g. from a Python content
|
|
builder), use it directly as sx source. If it evaluates to an AST/list,
|
|
serialize it to sx wire format via async_eval_to_sx.
|
|
"""
|
|
from .html import _RawHTML
|
|
from .parser import SxExpr
|
|
# First try async_eval to get the raw value
|
|
result = await async_eval_fn(expr, env, ctx)
|
|
# If it's already an sx source string, use as-is
|
|
if isinstance(result, str):
|
|
return result
|
|
if isinstance(result, _RawHTML):
|
|
return result.html
|
|
if isinstance(result, SxExpr):
|
|
return result.source
|
|
if result is None:
|
|
return ""
|
|
# For other types (lists, components rendered to HTML via _RawHTML, etc.),
|
|
# serialize to sx wire format
|
|
from .parser import serialize
|
|
return serialize(result)
|
|
|
|
|
|
async def execute_page(
|
|
page_def: PageDef,
|
|
service_name: str,
|
|
url_params: dict[str, Any] | None = None,
|
|
) -> str:
|
|
"""Execute a declarative page and return the full response string.
|
|
|
|
1. Build env from component env + page closure + URL params
|
|
2. If :data — async_eval(data_expr) → merge result dict into env
|
|
3. Render slots: async_eval_to_sx(content_expr) etc.
|
|
4. get_template_context() for header construction
|
|
5. Resolve layout → header rows
|
|
6. Branch: full_page_sx() vs oob_page_sx() based on is_htmx_request()
|
|
"""
|
|
from .jinja_bridge import get_component_env, _get_request_context
|
|
from .async_eval import async_eval, async_eval_to_sx
|
|
from .page import get_template_context
|
|
from .helpers import full_page_sx, oob_page_sx, sx_response
|
|
from .layouts import get_layout
|
|
from shared.browser.app.utils.htmx import is_htmx_request
|
|
|
|
if url_params is None:
|
|
url_params = {}
|
|
|
|
# Build environment
|
|
env = dict(get_component_env())
|
|
env.update(get_page_helpers(service_name))
|
|
env.update(page_def.closure)
|
|
|
|
# Inject URL params as kebab-case symbols
|
|
for key, val in url_params.items():
|
|
kebab = key.replace("_", "-")
|
|
env[kebab] = val
|
|
env[key] = val # also provide snake_case for convenience
|
|
|
|
# Get request context for I/O primitives
|
|
ctx = _get_request_context()
|
|
|
|
# Evaluate :data expression if present
|
|
if page_def.data_expr is not None:
|
|
data_result = await async_eval(page_def.data_expr, env, ctx)
|
|
if isinstance(data_result, dict):
|
|
env.update(data_result)
|
|
|
|
# Render content slot (required)
|
|
content_sx = await _eval_slot(page_def.content_expr, env, ctx, async_eval, async_eval_to_sx)
|
|
|
|
# Render optional slots
|
|
filter_sx = ""
|
|
if page_def.filter_expr is not None:
|
|
filter_sx = await _eval_slot(page_def.filter_expr, env, ctx, async_eval, async_eval_to_sx)
|
|
|
|
aside_sx = ""
|
|
if page_def.aside_expr is not None:
|
|
aside_sx = await _eval_slot(page_def.aside_expr, env, ctx, async_eval, async_eval_to_sx)
|
|
|
|
menu_sx = ""
|
|
if page_def.menu_expr is not None:
|
|
menu_sx = await _eval_slot(page_def.menu_expr, env, ctx, async_eval, async_eval_to_sx)
|
|
|
|
# Resolve layout → header rows
|
|
tctx = await get_template_context()
|
|
header_rows = ""
|
|
oob_headers = ""
|
|
layout_kwargs: dict[str, Any] = {}
|
|
|
|
if page_def.layout is not None:
|
|
if isinstance(page_def.layout, str):
|
|
layout_name = page_def.layout
|
|
elif isinstance(page_def.layout, list):
|
|
# (:layout-name :key val ...) — unevaluated list from defpage
|
|
# Evaluate each value in the current env to resolve dynamic bindings
|
|
from .types import Keyword as SxKeyword, Symbol as SxSymbol
|
|
raw = page_def.layout
|
|
# First element is layout name (keyword or symbol or string)
|
|
first = raw[0]
|
|
if isinstance(first, SxKeyword):
|
|
layout_name = first.name
|
|
elif isinstance(first, SxSymbol):
|
|
layout_name = first.name
|
|
elif isinstance(first, str):
|
|
layout_name = first
|
|
else:
|
|
layout_name = str(first)
|
|
# Parse keyword args — evaluate values at request time
|
|
i = 1
|
|
while i < len(raw):
|
|
k = raw[i]
|
|
if isinstance(k, SxKeyword) and i + 1 < len(raw):
|
|
raw_val = raw[i + 1]
|
|
resolved = await async_eval(raw_val, env, ctx)
|
|
layout_kwargs[k.name.replace("-", "_")] = resolved
|
|
i += 2
|
|
else:
|
|
i += 1
|
|
else:
|
|
layout_name = str(page_def.layout)
|
|
|
|
layout = get_layout(layout_name)
|
|
if layout is not None:
|
|
header_rows = await layout.full_headers(tctx, **layout_kwargs)
|
|
oob_headers = await layout.oob_headers(tctx, **layout_kwargs)
|
|
|
|
# Branch on request type
|
|
is_htmx = is_htmx_request()
|
|
|
|
if is_htmx:
|
|
return sx_response(oob_page_sx(
|
|
oobs=oob_headers if oob_headers else "",
|
|
filter=filter_sx,
|
|
aside=aside_sx,
|
|
content=content_sx,
|
|
menu=menu_sx,
|
|
))
|
|
else:
|
|
return full_page_sx(
|
|
tctx,
|
|
header_rows=header_rows,
|
|
filter=filter_sx,
|
|
aside=aside_sx,
|
|
content=content_sx,
|
|
menu=menu_sx,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Blueprint mounting
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def mount_pages(bp: Any, service_name: str) -> None:
|
|
"""Mount all registered PageDef routes onto a Quart Blueprint.
|
|
|
|
For each PageDef, adds a GET route with appropriate auth/cache
|
|
decorators. Coexists with existing Python routes on the same blueprint.
|
|
"""
|
|
from quart import make_response
|
|
|
|
pages = get_all_pages(service_name)
|
|
|
|
for page_def in pages.values():
|
|
_mount_one_page(bp, service_name, page_def)
|
|
|
|
|
|
def _mount_one_page(bp: Any, service_name: str, page_def: PageDef) -> None:
|
|
"""Mount a single PageDef as a GET route on the blueprint."""
|
|
from quart import make_response
|
|
|
|
# Build the view function
|
|
async def page_view(**kwargs: Any) -> Any:
|
|
# Re-fetch the page from registry to support hot-reload of content
|
|
current = get_page(service_name, page_def.name) or page_def
|
|
result = await execute_page(current, service_name, url_params=kwargs)
|
|
# If result is already a Response (from sx_response), return it
|
|
if hasattr(result, "status_code"):
|
|
return result
|
|
return await make_response(result, 200)
|
|
|
|
# Give the view function a unique name for Quart's routing
|
|
page_view.__name__ = f"defpage_{page_def.name.replace('-', '_')}"
|
|
page_view.__qualname__ = page_view.__name__
|
|
|
|
# Apply auth decorator
|
|
view_fn = _apply_auth(page_view, page_def.auth)
|
|
|
|
# Apply cache decorator
|
|
if page_def.cache:
|
|
view_fn = _apply_cache(view_fn, page_def.cache)
|
|
|
|
# Register the route
|
|
bp.add_url_rule(
|
|
page_def.path,
|
|
endpoint=page_view.__name__,
|
|
view_func=view_fn,
|
|
methods=["GET"],
|
|
)
|
|
logger.info("Mounted defpage %s:%s → GET %s", service_name, page_def.name, page_def.path)
|
|
|
|
|
|
def _apply_auth(fn: Any, auth: str | list) -> Any:
|
|
"""Wrap a view function with the appropriate auth decorator."""
|
|
if auth == "public":
|
|
return fn
|
|
if auth == "login":
|
|
from shared.browser.app.authz import require_login
|
|
return require_login(fn)
|
|
if auth == "admin":
|
|
from shared.browser.app.authz import require_admin
|
|
return require_admin(fn)
|
|
if isinstance(auth, list) and auth and auth[0] == "rights":
|
|
from shared.browser.app.authz import require_rights
|
|
return require_rights(*auth[1:])(fn)
|
|
return fn
|
|
|
|
|
|
def _apply_cache(fn: Any, cache: dict) -> Any:
|
|
"""Wrap a view function with cache_page decorator."""
|
|
from shared.browser.app.redis_cacher import cache_page
|
|
ttl = cache.get("ttl", 0)
|
|
tag = cache.get("tag")
|
|
scope = cache.get("scope", "user")
|
|
return cache_page(ttl=ttl, tag=tag, scope=scope)(fn)
|