giles
04366990ec
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 2m33s
Add boundary.sx declaring all 34 I/O primitives, 32 page helpers, and 9 allowed boundary types. Runtime validation in boundary.py checks every registration against the spec — undeclared primitives/helpers crash at startup with SX_BOUNDARY_STRICT=1 (now set in both dev and prod). Key changes: - Move 5 I/O-in-disguise primitives (app-url, asset-url, config, jinja-global, relations-from) from primitives.py to primitives_io.py - Remove duplicate url-for/route-prefix from primitives.py (already in IO) - Fix parse-datetime to return ISO string instead of raw datetime - Add datetime→isoformat conversion in _convert_result at the edge - Wrap page helper return values with boundary type validation - Replace all SxExpr(f"...") patterns with sx_call() or _sx_fragment() - Add assert declaration to primitives.sx Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1061 lines
35 KiB
Python
1061 lines
35 KiB
Python
"""
|
|
Async I/O primitives for the s-expression resolver.
|
|
|
|
These wrap rose-ash's inter-service communication layer so that s-expressions
|
|
can fetch fragments, query data, call actions, and access request context.
|
|
|
|
Unlike pure primitives (primitives.py), these are **async** and are executed
|
|
by the resolver rather than the evaluator. They are identified by name
|
|
during the tree-walk phase and dispatched via ``asyncio.gather()``.
|
|
|
|
Usage in s-expressions::
|
|
|
|
(frag "blog" "link-card" :slug "apple")
|
|
(query "market" "products-by-ids" :ids "1,2,3")
|
|
(action "market" "create-marketplace" :name "Farm Shop" :slug "farm")
|
|
(current-user)
|
|
(htmx-request?)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextvars
|
|
from typing import Any
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry of async primitives (name → metadata)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Names that the resolver recognises as I/O nodes requiring async resolution.
|
|
# The resolver collects these during tree-walk, groups them, and dispatches
|
|
# them in parallel.
|
|
IO_PRIMITIVES: frozenset[str] = frozenset({
|
|
"frag",
|
|
"query",
|
|
"action",
|
|
"current-user",
|
|
"htmx-request?",
|
|
"service",
|
|
"request-arg",
|
|
"request-path",
|
|
"nav-tree",
|
|
"get-children",
|
|
"g",
|
|
"csrf-token",
|
|
"abort",
|
|
"url-for",
|
|
"route-prefix",
|
|
"root-header-ctx",
|
|
"post-header-ctx",
|
|
"select-colours",
|
|
"account-nav-ctx",
|
|
"app-rights",
|
|
"federation-actor-ctx",
|
|
"request-view-args",
|
|
"cart-page-ctx",
|
|
"events-calendar-ctx",
|
|
"events-day-ctx",
|
|
"events-entry-ctx",
|
|
"events-slot-ctx",
|
|
"events-ticket-type-ctx",
|
|
"market-header-ctx",
|
|
"app-url",
|
|
"asset-url",
|
|
"config",
|
|
"jinja-global",
|
|
"relations-from",
|
|
})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Request context (set per-request by the resolver)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ContextVar for the handler's domain service object.
|
|
# Set by the handler blueprint before executing a defhandler.
|
|
_handler_service: contextvars.ContextVar[Any] = contextvars.ContextVar(
|
|
"_handler_service", default=None
|
|
)
|
|
|
|
|
|
def set_handler_service(service_obj: Any) -> None:
|
|
"""Bind the local domain service for ``(service ...)`` primitive calls."""
|
|
_handler_service.set(service_obj)
|
|
|
|
|
|
def get_handler_service() -> Any:
|
|
"""Get the currently bound handler service, or None."""
|
|
return _handler_service.get(None)
|
|
|
|
|
|
class RequestContext:
|
|
"""Per-request context provided to I/O primitives.
|
|
|
|
Populated by the resolver from the Quart request before resolution begins.
|
|
"""
|
|
__slots__ = ("user", "is_htmx", "extras")
|
|
|
|
def __init__(
|
|
self,
|
|
user: dict[str, Any] | None = None,
|
|
is_htmx: bool = False,
|
|
extras: dict[str, Any] | None = None,
|
|
):
|
|
self.user = user
|
|
self.is_htmx = is_htmx
|
|
self.extras = extras or {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# I/O dispatch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def execute_io(
|
|
name: str,
|
|
args: list[Any],
|
|
kwargs: dict[str, Any],
|
|
ctx: RequestContext,
|
|
) -> Any:
|
|
"""Execute an I/O primitive by name.
|
|
|
|
Called by the resolver after collecting and grouping I/O nodes.
|
|
Returns the result to be substituted back into the tree.
|
|
"""
|
|
handler = _IO_HANDLERS.get(name)
|
|
if handler is None:
|
|
raise RuntimeError(f"Unknown I/O primitive: {name}")
|
|
return await handler(args, kwargs, ctx)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Individual handlers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _clean_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]:
|
|
"""Strip None and NIL values from kwargs for Python interop."""
|
|
from .types import NIL
|
|
return {k: v for k, v in kwargs.items() if v is not None and v is not NIL}
|
|
|
|
|
|
async def _io_frag(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> str:
|
|
"""``(frag "service" "type" :key val ...)`` → fetch_fragment."""
|
|
if len(args) < 2:
|
|
raise ValueError("frag requires service and fragment type")
|
|
service = str(args[0])
|
|
frag_type = str(args[1])
|
|
params = _clean_kwargs(kwargs)
|
|
|
|
from shared.infrastructure.fragments import fetch_fragment
|
|
return await fetch_fragment(service, frag_type, params=params or None)
|
|
|
|
|
|
async def _io_query(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> Any:
|
|
"""``(query "service" "query-name" :key val ...)`` → fetch_data."""
|
|
if len(args) < 2:
|
|
raise ValueError("query requires service and query name")
|
|
service = str(args[0])
|
|
query_name = str(args[1])
|
|
params = _clean_kwargs(kwargs)
|
|
|
|
from shared.infrastructure.data_client import fetch_data
|
|
return await fetch_data(service, query_name, params=params or None)
|
|
|
|
|
|
async def _io_action(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> Any:
|
|
"""``(action "service" "action-name" :key val ...)`` → call_action."""
|
|
if len(args) < 2:
|
|
raise ValueError("action requires service and action name")
|
|
service = str(args[0])
|
|
action_name = str(args[1])
|
|
payload = _clean_kwargs(kwargs)
|
|
|
|
from shared.infrastructure.actions import call_action
|
|
return await call_action(service, action_name, payload=payload or None)
|
|
|
|
|
|
async def _io_current_user(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any] | None:
|
|
"""``(current-user)`` → user dict from request context."""
|
|
return ctx.user
|
|
|
|
|
|
async def _io_htmx_request(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> bool:
|
|
"""``(htmx-request?)`` → True if HX-Request header present."""
|
|
return ctx.is_htmx
|
|
|
|
|
|
async def _io_service(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> Any:
|
|
"""``(service "svc-name" "method-name" :key val ...)`` → call domain service.
|
|
|
|
Looks up the service from the shared registry by name, then calls the
|
|
named method with ``g.s`` (async session) + keyword args. Falls back
|
|
to the bound handler service if only one positional arg is given.
|
|
"""
|
|
if not args:
|
|
raise ValueError("service requires at least a method name")
|
|
|
|
if len(args) >= 2:
|
|
# (service "calendar" "associated-entries" :key val ...)
|
|
from shared.services.registry import services as svc_registry
|
|
svc_name = str(args[0]).replace("-", "_")
|
|
svc = getattr(svc_registry, svc_name, None)
|
|
if svc is None:
|
|
raise RuntimeError(f"No service registered as: {svc_name}")
|
|
method_name = str(args[1]).replace("-", "_")
|
|
else:
|
|
# (service "method-name" :key val ...) — legacy / bound service
|
|
svc = get_handler_service()
|
|
if svc is None:
|
|
raise RuntimeError(
|
|
"No handler service bound — cannot call (service ...)")
|
|
method_name = str(args[0]).replace("-", "_")
|
|
|
|
method = getattr(svc, method_name, None)
|
|
if method is None:
|
|
raise RuntimeError(f"Service has no method: {method_name}")
|
|
|
|
# Convert kwarg keys from kebab-case to snake_case, NIL → None
|
|
from .types import NIL
|
|
clean_kwargs = {
|
|
k.replace("-", "_"): (None if v is NIL else v)
|
|
for k, v in kwargs.items()
|
|
}
|
|
from quart import g
|
|
result = await method(g.s, **clean_kwargs)
|
|
|
|
return _convert_result(result)
|
|
|
|
|
|
def _dto_to_dict(obj: Any) -> dict[str, Any]:
|
|
"""Convert a DTO/dataclass/namedtuple to a plain dict.
|
|
|
|
Adds ``{field}_year``, ``{field}_month``, ``{field}_day`` convenience
|
|
keys for any datetime-valued field so sx handlers can build URL paths
|
|
without parsing date strings.
|
|
"""
|
|
if hasattr(obj, "__dataclass_fields__"):
|
|
from shared.contracts.dtos import dto_to_dict
|
|
return dto_to_dict(obj)
|
|
elif hasattr(obj, "_asdict"):
|
|
d = dict(obj._asdict())
|
|
elif hasattr(obj, "__dict__"):
|
|
d = {k: v for k, v in obj.__dict__.items() if not k.startswith("_")}
|
|
else:
|
|
return {"value": obj}
|
|
# Expand datetime fields into year/month/day convenience keys
|
|
for key, val in list(d.items()):
|
|
if hasattr(val, "year") and hasattr(val, "strftime"):
|
|
d[f"{key}_year"] = val.year
|
|
d[f"{key}_month"] = val.month
|
|
d[f"{key}_day"] = val.day
|
|
return d
|
|
|
|
|
|
def _convert_result(result: Any) -> Any:
|
|
"""Convert a service method result for sx consumption.
|
|
|
|
Converts DTOs/dataclasses to dicts, datetimes to ISO strings,
|
|
and ensures only SX-typed values cross the boundary.
|
|
"""
|
|
if result is None:
|
|
from .types import NIL
|
|
return NIL
|
|
if isinstance(result, (int, float, str, bool)):
|
|
return result
|
|
# datetime → ISO string at the edge
|
|
if hasattr(result, "isoformat") and callable(result.isoformat):
|
|
return result.isoformat()
|
|
if isinstance(result, dict):
|
|
return {k: _convert_result(v) for k, v in result.items()}
|
|
if isinstance(result, tuple):
|
|
# Tuple returns (e.g. (entries, has_more)) → list for sx access
|
|
return [_convert_result(item) for item in result]
|
|
if hasattr(result, "__dataclass_fields__") or hasattr(result, "_asdict"):
|
|
return _dto_to_dict(result)
|
|
if isinstance(result, list):
|
|
return [
|
|
_dto_to_dict(item)
|
|
if hasattr(item, "__dataclass_fields__") or hasattr(item, "_asdict")
|
|
else _convert_result(item)
|
|
for item in result
|
|
]
|
|
return result
|
|
|
|
|
|
async def _io_request_arg(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> Any:
|
|
"""``(request-arg "name" default?)`` → request.args.get(name, default)."""
|
|
if not args:
|
|
raise ValueError("request-arg requires a name")
|
|
from quart import request
|
|
name = str(args[0])
|
|
default = args[1] if len(args) > 1 else None
|
|
return request.args.get(name, default)
|
|
|
|
|
|
async def _io_request_path(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> str:
|
|
"""``(request-path)`` → request.path."""
|
|
from quart import request
|
|
return request.path
|
|
|
|
|
|
async def _io_nav_tree(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> list[dict[str, Any]]:
|
|
"""``(nav-tree)`` → list of navigation menu node dicts."""
|
|
from quart import g
|
|
from shared.services.navigation import get_navigation_tree
|
|
nodes = await get_navigation_tree(g.s)
|
|
return [_dto_to_dict(node) for node in nodes]
|
|
|
|
|
|
async def _io_get_children(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> list[dict[str, Any]]:
|
|
"""``(get-children :parent-type "page" :parent-id 1 ...)``"""
|
|
from quart import g
|
|
from shared.services.relationships import get_children
|
|
clean = {k.replace("-", "_"): v for k, v in kwargs.items()}
|
|
children = await get_children(g.s, **clean)
|
|
return [_dto_to_dict(child) for child in children]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Handler registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _io_g(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> Any:
|
|
"""``(g "key")`` → getattr(g, key, None).
|
|
|
|
Reads a value from the Quart request-local ``g`` object.
|
|
Kebab-case keys are converted to snake_case automatically.
|
|
"""
|
|
from quart import g
|
|
key = str(args[0]).replace("-", "_") if args else ""
|
|
return getattr(g, key, None)
|
|
|
|
|
|
async def _io_csrf_token(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> str:
|
|
"""``(csrf-token)`` → current CSRF token string."""
|
|
from quart import current_app
|
|
csrf = current_app.jinja_env.globals.get("csrf_token")
|
|
if callable(csrf):
|
|
return csrf()
|
|
return ""
|
|
|
|
|
|
async def _io_abort(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> Any:
|
|
"""``(abort 403 "message")`` — raise HTTP error from SX.
|
|
|
|
Allows defpages to abort with HTTP error codes for auth/ownership
|
|
checks without needing a Python page helper.
|
|
"""
|
|
if not args:
|
|
raise ValueError("abort requires a status code")
|
|
from quart import abort
|
|
status = int(args[0])
|
|
message = str(args[1]) if len(args) > 1 else ""
|
|
abort(status, message)
|
|
|
|
|
|
async def _io_url_for(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> str:
|
|
"""``(url-for "endpoint" :key val ...)`` → url_for(endpoint, **kwargs).
|
|
|
|
Generates a URL for the given endpoint. Keyword args become URL
|
|
parameters (kebab-case converted to snake_case).
|
|
"""
|
|
if not args:
|
|
raise ValueError("url-for requires an endpoint name")
|
|
from quart import url_for
|
|
endpoint = str(args[0])
|
|
clean = {k.replace("-", "_"): v for k, v in _clean_kwargs(kwargs).items()}
|
|
# Convert numeric values for int URL params
|
|
for k, v in clean.items():
|
|
if isinstance(v, str) and v.isdigit():
|
|
clean[k] = int(v)
|
|
return url_for(endpoint, **clean)
|
|
|
|
|
|
async def _io_route_prefix(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> str:
|
|
"""``(route-prefix)`` → current route prefix string."""
|
|
from shared.utils import route_prefix
|
|
return route_prefix()
|
|
|
|
|
|
async def _io_root_header_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any]:
|
|
"""``(root-header-ctx)`` → dict with all root header values.
|
|
|
|
Fetches cart-mini, auth-menu, nav-tree fragments and computes
|
|
settings-url / is-admin from rights. Result is cached on ``g``
|
|
per request so multiple calls (e.g. header + mobile) are free.
|
|
"""
|
|
from quart import g, current_app, request
|
|
cached = getattr(g, "_root_header_ctx", None)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
from shared.infrastructure.fragments import fetch_fragments
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
from shared.infrastructure.urls import app_url
|
|
from shared.config import config
|
|
from .types import NIL
|
|
|
|
user = getattr(g, "user", None)
|
|
ident = current_cart_identity()
|
|
|
|
cart_params: dict[str, Any] = {}
|
|
if ident["user_id"] is not None:
|
|
cart_params["user_id"] = ident["user_id"]
|
|
if ident["session_id"] is not None:
|
|
cart_params["session_id"] = ident["session_id"]
|
|
|
|
auth_params: dict[str, Any] = {}
|
|
if user and getattr(user, "email", None):
|
|
auth_params["email"] = user.email
|
|
|
|
nav_params = {"app_name": current_app.name, "path": request.path}
|
|
|
|
cart_mini, auth_menu, nav_tree = await fetch_fragments([
|
|
("cart", "cart-mini", cart_params or None),
|
|
("account", "auth-menu", auth_params or None),
|
|
("blog", "nav-tree", nav_params),
|
|
])
|
|
|
|
rights = getattr(g, "rights", None) or {}
|
|
is_admin = (
|
|
rights.get("admin", False)
|
|
if isinstance(rights, dict)
|
|
else getattr(rights, "admin", False)
|
|
)
|
|
|
|
result = {
|
|
"cart-mini": cart_mini or NIL,
|
|
"blog-url": app_url("blog", ""),
|
|
"site-title": config()["title"],
|
|
"app-label": current_app.name,
|
|
"nav-tree": nav_tree or NIL,
|
|
"auth-menu": auth_menu or NIL,
|
|
"nav-panel": NIL,
|
|
"settings-url": app_url("blog", "/settings/") if is_admin else "",
|
|
"is-admin": is_admin,
|
|
}
|
|
g._root_header_ctx = result
|
|
return result
|
|
|
|
|
|
async def _io_select_colours(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> str:
|
|
"""``(select-colours)`` → the shared select/hover CSS class string."""
|
|
from quart import current_app
|
|
return current_app.jinja_env.globals.get("select_colours", "")
|
|
|
|
|
|
async def _io_account_nav_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> Any:
|
|
"""``(account-nav-ctx)`` → account nav fragments as SxExpr, or NIL.
|
|
|
|
Reads ``g.account_nav`` (set by account service's before_request hook),
|
|
wrapping HTML strings in ``~rich-text`` for SX rendering.
|
|
"""
|
|
from quart import g
|
|
from .types import NIL
|
|
from .parser import SxExpr
|
|
from .helpers import sx_call
|
|
val = getattr(g, "account_nav", None)
|
|
if not val:
|
|
return NIL
|
|
if isinstance(val, SxExpr):
|
|
return val
|
|
# HTML string → wrap for SX rendering
|
|
return sx_call("rich-text", html=str(val))
|
|
|
|
|
|
async def _io_app_rights(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any]:
|
|
"""``(app-rights)`` → user rights dict from ``g.rights``."""
|
|
from quart import g
|
|
return getattr(g, "rights", None) or {}
|
|
|
|
|
|
async def _io_post_header_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any]:
|
|
"""``(post-header-ctx)`` → dict with post-level header values.
|
|
|
|
Reads post data from ``g._defpage_ctx`` (set by per-service page
|
|
helpers), fetches container-nav and page cart count. Result is
|
|
cached on ``g`` per request.
|
|
|
|
Returns dict with keys: slug, title, feature-image, link-href,
|
|
container-nav, page-cart-count, cart-href, admin-href, is-admin,
|
|
is-admin-page, select-colours.
|
|
"""
|
|
from quart import g, request
|
|
cached = getattr(g, "_post_header_ctx", None)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
from shared.infrastructure.urls import app_url
|
|
from .types import NIL
|
|
from .parser import SxExpr
|
|
|
|
dctx = getattr(g, "_defpage_ctx", None) or {}
|
|
post = dctx.get("post") or {}
|
|
slug = post.get("slug", "")
|
|
if not slug:
|
|
result: dict[str, Any] = {"slug": ""}
|
|
g._post_header_ctx = result
|
|
return result
|
|
|
|
title = (post.get("title") or "")[:160]
|
|
feature_image = post.get("feature_image") or NIL
|
|
|
|
# Container nav (pre-fetched by page helper into defpage ctx)
|
|
raw_nav = dctx.get("container_nav") or ""
|
|
container_nav: Any = NIL
|
|
nav_str = str(raw_nav).strip()
|
|
if nav_str and nav_str.replace("(<>", "").replace(")", "").strip():
|
|
if isinstance(raw_nav, SxExpr):
|
|
container_nav = raw_nav
|
|
else:
|
|
container_nav = SxExpr(nav_str)
|
|
|
|
page_cart_count = dctx.get("page_cart_count", 0) or 0
|
|
|
|
rights = getattr(g, "rights", None) or {}
|
|
is_admin = (
|
|
rights.get("admin", False)
|
|
if isinstance(rights, dict)
|
|
else getattr(rights, "admin", False)
|
|
)
|
|
|
|
is_admin_page = dctx.get("is_admin_section") or "/admin" in request.path
|
|
|
|
from quart import current_app
|
|
select_colours = current_app.jinja_env.globals.get("select_colours", "")
|
|
|
|
result = {
|
|
"slug": slug,
|
|
"title": title,
|
|
"feature-image": feature_image,
|
|
"link-href": app_url("blog", f"/{slug}/"),
|
|
"container-nav": container_nav,
|
|
"page-cart-count": page_cart_count,
|
|
"cart-href": app_url("cart", f"/{slug}/") if page_cart_count else "",
|
|
"admin-href": app_url("blog", f"/{slug}/admin/"),
|
|
"is-admin": is_admin,
|
|
"is-admin-page": is_admin_page or NIL,
|
|
"select-colours": select_colours,
|
|
}
|
|
g._post_header_ctx = result
|
|
return result
|
|
|
|
|
|
async def _io_cart_page_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any]:
|
|
"""``(cart-page-ctx)`` → dict with cart page header values.
|
|
|
|
Reads ``g.page_post`` (set by cart's before_request) and returns
|
|
slug, title, feature-image, and cart-url for the page cart header.
|
|
"""
|
|
from quart import g
|
|
from .types import NIL
|
|
from shared.infrastructure.urls import app_url
|
|
|
|
page_post = getattr(g, "page_post", None)
|
|
if not page_post:
|
|
return {"slug": "", "title": "", "feature-image": NIL, "cart-url": "/"}
|
|
|
|
slug = getattr(page_post, "slug", "") or ""
|
|
title = (getattr(page_post, "title", "") or "")[:160]
|
|
feature_image = getattr(page_post, "feature_image", None) or NIL
|
|
|
|
return {
|
|
"slug": slug,
|
|
"title": title,
|
|
"feature-image": feature_image,
|
|
"page-cart-url": app_url("cart", f"/{slug}/"),
|
|
"cart-url": app_url("cart", "/"),
|
|
}
|
|
|
|
|
|
async def _io_federation_actor_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any] | None:
|
|
"""``(federation-actor-ctx)`` → serialized actor dict or None.
|
|
|
|
Reads ``g._social_actor`` (set by federation social blueprint's
|
|
before_request hook) and serializes to a dict for .sx components.
|
|
"""
|
|
from quart import g
|
|
actor = getattr(g, "_social_actor", None)
|
|
if not actor:
|
|
return None
|
|
return {
|
|
"id": actor.id,
|
|
"preferred_username": actor.preferred_username,
|
|
"display_name": getattr(actor, "display_name", None),
|
|
"icon_url": getattr(actor, "icon_url", None),
|
|
"actor_url": getattr(actor, "actor_url", ""),
|
|
}
|
|
|
|
|
|
async def _io_request_view_args(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> Any:
|
|
"""``(request-view-args "key")`` → request.view_args[key]."""
|
|
if not args:
|
|
raise ValueError("request-view-args requires a key")
|
|
from quart import request
|
|
key = str(args[0])
|
|
return (request.view_args or {}).get(key)
|
|
|
|
|
|
async def _io_events_calendar_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any]:
|
|
"""``(events-calendar-ctx)`` → dict with events calendar header values.
|
|
|
|
Reads ``g.calendar`` or ``g._defpage_ctx["calendar"]`` and returns
|
|
slug, name, description for the calendar header row.
|
|
"""
|
|
from quart import g
|
|
cal = getattr(g, "calendar", None)
|
|
if not cal:
|
|
dctx = getattr(g, "_defpage_ctx", None) or {}
|
|
cal = dctx.get("calendar")
|
|
if not cal:
|
|
return {"slug": ""}
|
|
return {
|
|
"slug": getattr(cal, "slug", "") or "",
|
|
"name": getattr(cal, "name", "") or "",
|
|
"description": getattr(cal, "description", "") or "",
|
|
}
|
|
|
|
|
|
async def _io_events_day_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any]:
|
|
"""``(events-day-ctx)`` → dict with events day header values.
|
|
|
|
Reads ``g.day_date``, ``g.calendar``, confirmed entries from
|
|
``g._defpage_ctx``. Pre-builds the confirmed entries nav as SxExpr.
|
|
"""
|
|
from quart import g, url_for
|
|
from .types import NIL
|
|
from .parser import SxExpr
|
|
|
|
dctx = getattr(g, "_defpage_ctx", None) or {}
|
|
cal = getattr(g, "calendar", None) or dctx.get("calendar")
|
|
day_date = dctx.get("day_date") or getattr(g, "day_date", None)
|
|
if not cal or not day_date:
|
|
return {"date-str": ""}
|
|
|
|
cal_slug = getattr(cal, "slug", "") or ""
|
|
|
|
# Build confirmed entries nav
|
|
confirmed = dctx.get("confirmed_entries") or []
|
|
rights = getattr(g, "rights", None) or {}
|
|
is_admin = (
|
|
rights.get("admin", False)
|
|
if isinstance(rights, dict)
|
|
else getattr(rights, "admin", False)
|
|
)
|
|
|
|
from .helpers import sx_call
|
|
nav_parts: list[str] = []
|
|
if confirmed:
|
|
entry_links = []
|
|
for entry in confirmed:
|
|
href = url_for(
|
|
"calendar.day.calendar_entries.calendar_entry.get",
|
|
calendar_slug=cal_slug,
|
|
year=day_date.year, month=day_date.month, day=day_date.day,
|
|
entry_id=entry.id,
|
|
)
|
|
start = entry.start_at.strftime("%H:%M") if entry.start_at else ""
|
|
end = (
|
|
f" \u2013 {entry.end_at.strftime('%H:%M')}"
|
|
if entry.end_at else ""
|
|
)
|
|
entry_links.append(sx_call(
|
|
"events-day-entry-link",
|
|
href=href, name=entry.name, time_str=f"{start}{end}",
|
|
))
|
|
inner = "".join(entry_links)
|
|
nav_parts.append(sx_call(
|
|
"events-day-entries-nav", inner=SxExpr(inner),
|
|
))
|
|
|
|
if is_admin and day_date:
|
|
admin_href = url_for(
|
|
"defpage_day_admin", calendar_slug=cal_slug,
|
|
year=day_date.year, month=day_date.month, day=day_date.day,
|
|
)
|
|
nav_parts.append(sx_call("nav-link", href=admin_href, icon="fa fa-cog"))
|
|
|
|
return {
|
|
"date-str": day_date.strftime("%A %d %B %Y"),
|
|
"year": day_date.year,
|
|
"month": day_date.month,
|
|
"day": day_date.day,
|
|
"nav": SxExpr("".join(nav_parts)) if nav_parts else NIL,
|
|
}
|
|
|
|
|
|
async def _io_events_entry_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any]:
|
|
"""``(events-entry-ctx)`` → dict with events entry header values.
|
|
|
|
Reads ``g.entry``, ``g.calendar``, and entry_posts from
|
|
``g._defpage_ctx``. Pre-builds entry nav (posts + admin link) as SxExpr.
|
|
"""
|
|
from quart import g, url_for
|
|
from .types import NIL
|
|
from .parser import SxExpr
|
|
|
|
dctx = getattr(g, "_defpage_ctx", None) or {}
|
|
cal = getattr(g, "calendar", None) or dctx.get("calendar")
|
|
entry = getattr(g, "entry", None) or dctx.get("entry")
|
|
if not cal or not entry:
|
|
return {"id": ""}
|
|
|
|
cal_slug = getattr(cal, "slug", "") or ""
|
|
day = dctx.get("day")
|
|
month = dctx.get("month")
|
|
year = dctx.get("year")
|
|
|
|
# Times
|
|
start = entry.start_at
|
|
end = entry.end_at
|
|
time_str = ""
|
|
if start:
|
|
time_str = start.strftime("%H:%M")
|
|
if end:
|
|
time_str += f" \u2192 {end.strftime('%H:%M')}"
|
|
|
|
link_href = url_for(
|
|
"calendar.day.calendar_entries.calendar_entry.get",
|
|
calendar_slug=cal_slug,
|
|
year=year, month=month, day=day, entry_id=entry.id,
|
|
)
|
|
|
|
# Build nav: associated posts + admin link
|
|
entry_posts = dctx.get("entry_posts") or []
|
|
rights = getattr(g, "rights", None) or {}
|
|
is_admin = (
|
|
rights.get("admin", False)
|
|
if isinstance(rights, dict)
|
|
else getattr(rights, "admin", False)
|
|
)
|
|
|
|
from .helpers import sx_call
|
|
from shared.infrastructure.urls import app_url
|
|
|
|
nav_parts: list[str] = []
|
|
if entry_posts:
|
|
post_links = ""
|
|
for ep in entry_posts:
|
|
ep_slug = getattr(ep, "slug", "")
|
|
ep_title = getattr(ep, "title", "")
|
|
feat = getattr(ep, "feature_image", None)
|
|
href = app_url("blog", f"/{ep_slug}/")
|
|
if feat:
|
|
img_html = sx_call("events-post-img", src=feat, alt=ep_title)
|
|
else:
|
|
img_html = sx_call("events-post-img-placeholder")
|
|
post_links += sx_call(
|
|
"events-entry-nav-post-link",
|
|
href=href, img=SxExpr(img_html), title=ep_title,
|
|
)
|
|
nav_parts.append(
|
|
sx_call("events-entry-posts-nav-oob", items=SxExpr(post_links))
|
|
.replace(' :hx-swap-oob "true"', '')
|
|
)
|
|
|
|
if is_admin:
|
|
admin_url = url_for(
|
|
"calendar.day.calendar_entries.calendar_entry.admin.admin",
|
|
calendar_slug=cal_slug,
|
|
day=day, month=month, year=year, entry_id=entry.id,
|
|
)
|
|
nav_parts.append(sx_call("events-entry-admin-link", href=admin_url))
|
|
|
|
# Entry admin nav (ticket_types link)
|
|
admin_href = url_for(
|
|
"calendar.day.calendar_entries.calendar_entry.admin.admin",
|
|
calendar_slug=cal_slug,
|
|
day=day, month=month, year=year, entry_id=entry.id,
|
|
) if is_admin else ""
|
|
|
|
ticket_types_href = url_for(
|
|
"calendar.day.calendar_entries.calendar_entry.ticket_types.get",
|
|
calendar_slug=cal_slug, entry_id=entry.id,
|
|
year=year, month=month, day=day,
|
|
)
|
|
|
|
from quart import current_app
|
|
select_colours = current_app.jinja_env.globals.get("select_colours", "")
|
|
|
|
return {
|
|
"id": str(entry.id),
|
|
"name": entry.name or "",
|
|
"time-str": time_str,
|
|
"link-href": link_href,
|
|
"nav": SxExpr("".join(nav_parts)) if nav_parts else NIL,
|
|
"admin-href": admin_href,
|
|
"ticket-types-href": ticket_types_href,
|
|
"is-admin": is_admin,
|
|
"select-colours": select_colours,
|
|
}
|
|
|
|
|
|
async def _io_events_slot_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any]:
|
|
"""``(events-slot-ctx)`` → dict with events slot header values."""
|
|
from quart import g
|
|
dctx = getattr(g, "_defpage_ctx", None) or {}
|
|
slot = getattr(g, "slot", None) or dctx.get("slot")
|
|
if not slot:
|
|
return {"name": ""}
|
|
return {
|
|
"name": getattr(slot, "name", "") or "",
|
|
"description": getattr(slot, "description", "") or "",
|
|
}
|
|
|
|
|
|
async def _io_events_ticket_type_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any]:
|
|
"""``(events-ticket-type-ctx)`` → dict with ticket type header values."""
|
|
from quart import g, url_for
|
|
|
|
dctx = getattr(g, "_defpage_ctx", None) or {}
|
|
cal = getattr(g, "calendar", None) or dctx.get("calendar")
|
|
entry = getattr(g, "entry", None) or dctx.get("entry")
|
|
ticket_type = getattr(g, "ticket_type", None) or dctx.get("ticket_type")
|
|
if not cal or not entry or not ticket_type:
|
|
return {"id": ""}
|
|
|
|
cal_slug = getattr(cal, "slug", "") or ""
|
|
day = dctx.get("day")
|
|
month = dctx.get("month")
|
|
year = dctx.get("year")
|
|
|
|
link_href = url_for(
|
|
"calendar.day.calendar_entries.calendar_entry.ticket_types.ticket_type.get",
|
|
calendar_slug=cal_slug, year=year, month=month, day=day,
|
|
entry_id=entry.id, ticket_type_id=ticket_type.id,
|
|
)
|
|
|
|
return {
|
|
"id": str(ticket_type.id),
|
|
"name": getattr(ticket_type, "name", "") or "",
|
|
"link-href": link_href,
|
|
}
|
|
|
|
|
|
async def _io_app_url(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> str:
|
|
"""``(app-url "blog" "/my-post/")`` → full URL for service."""
|
|
if not args:
|
|
raise ValueError("app-url requires a service name")
|
|
from shared.infrastructure.urls import app_url
|
|
service = str(args[0])
|
|
path = str(args[1]) if len(args) > 1 else "/"
|
|
return app_url(service, path)
|
|
|
|
|
|
async def _io_asset_url(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> str:
|
|
"""``(asset-url "/img/logo.png")`` → versioned static URL."""
|
|
from shared.infrastructure.urls import asset_url
|
|
path = str(args[0]) if args else ""
|
|
return asset_url(path)
|
|
|
|
|
|
async def _io_config(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> Any:
|
|
"""``(config "key")`` → shared.config.config()[key]."""
|
|
if not args:
|
|
raise ValueError("config requires a key")
|
|
from shared.config import config
|
|
cfg = config()
|
|
return cfg.get(str(args[0]))
|
|
|
|
|
|
async def _io_jinja_global(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> Any:
|
|
"""``(jinja-global "key")`` → current_app.jinja_env.globals[key]."""
|
|
if not args:
|
|
raise ValueError("jinja-global requires a key")
|
|
from quart import current_app
|
|
key = str(args[0])
|
|
default = args[1] if len(args) > 1 else None
|
|
return current_app.jinja_env.globals.get(key, default)
|
|
|
|
|
|
async def _io_relations_from(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> list[dict]:
|
|
"""``(relations-from "page")`` → list of RelationDef dicts."""
|
|
if not args:
|
|
raise ValueError("relations-from requires an entity type")
|
|
from shared.sx.relations import relations_from
|
|
return [
|
|
{
|
|
"name": d.name, "from_type": d.from_type, "to_type": d.to_type,
|
|
"cardinality": d.cardinality, "nav": d.nav,
|
|
"nav_icon": d.nav_icon, "nav_label": d.nav_label,
|
|
}
|
|
for d in relations_from(str(args[0]))
|
|
]
|
|
|
|
|
|
async def _io_market_header_ctx(
|
|
args: list[Any], kwargs: dict[str, Any], ctx: RequestContext
|
|
) -> dict[str, Any]:
|
|
"""``(market-header-ctx)`` → dict with market header data.
|
|
|
|
Returns plain data (categories list, hrefs, flags) for the
|
|
~market-header-auto macro. Mobile nav is pre-built as SxExpr.
|
|
"""
|
|
from quart import g, url_for
|
|
from shared.config import config as get_config
|
|
from .parser import SxExpr
|
|
|
|
cfg = get_config()
|
|
market_title = cfg.get("market_title", "")
|
|
link_href = url_for("defpage_market_home")
|
|
|
|
# Get categories if market is loaded
|
|
market = getattr(g, "market", None)
|
|
categories = {}
|
|
if market:
|
|
from bp.browse.services.nav import get_nav
|
|
nav_data = await get_nav(g.s, market_id=market.id)
|
|
categories = nav_data.get("cats", {})
|
|
|
|
# Build minimal ctx for existing helper functions
|
|
select_colours = getattr(g, "select_colours", "")
|
|
if not select_colours:
|
|
from quart import current_app
|
|
select_colours = current_app.jinja_env.globals.get("select_colours", "")
|
|
rights = getattr(g, "rights", None) or {}
|
|
|
|
mini_ctx: dict[str, Any] = {
|
|
"market_title": market_title,
|
|
"top_slug": "",
|
|
"sub_slug": "",
|
|
"categories": categories,
|
|
"qs": "",
|
|
"hx_select_search": "#main-panel",
|
|
"select_colours": select_colours,
|
|
"rights": rights,
|
|
"category_label": "",
|
|
}
|
|
|
|
# Build header + mobile nav data via new data-driven helpers
|
|
from sxc.pages.layouts import _market_header_data, _mobile_nav_panel_sx
|
|
header_data = _market_header_data(mini_ctx)
|
|
mobile_nav = _mobile_nav_panel_sx(mini_ctx)
|
|
|
|
return {
|
|
"market-title": market_title,
|
|
"link-href": link_href,
|
|
"top-slug": "",
|
|
"sub-slug": "",
|
|
"categories": header_data.get("categories", []),
|
|
"hx-select": header_data.get("hx-select", "#main-panel"),
|
|
"select-colours": header_data.get("select-colours", ""),
|
|
"all-href": header_data.get("all-href", ""),
|
|
"all-active": header_data.get("all-active", False),
|
|
"admin-href": header_data.get("admin-href", ""),
|
|
"mobile-nav": SxExpr(mobile_nav) if mobile_nav else "",
|
|
}
|
|
|
|
|
|
_IO_HANDLERS: dict[str, Any] = {
|
|
"frag": _io_frag,
|
|
"query": _io_query,
|
|
"action": _io_action,
|
|
"current-user": _io_current_user,
|
|
"htmx-request?": _io_htmx_request,
|
|
"service": _io_service,
|
|
"request-arg": _io_request_arg,
|
|
"request-path": _io_request_path,
|
|
"nav-tree": _io_nav_tree,
|
|
"get-children": _io_get_children,
|
|
"g": _io_g,
|
|
"csrf-token": _io_csrf_token,
|
|
"abort": _io_abort,
|
|
"url-for": _io_url_for,
|
|
"route-prefix": _io_route_prefix,
|
|
"root-header-ctx": _io_root_header_ctx,
|
|
"post-header-ctx": _io_post_header_ctx,
|
|
"select-colours": _io_select_colours,
|
|
"account-nav-ctx": _io_account_nav_ctx,
|
|
"app-rights": _io_app_rights,
|
|
"federation-actor-ctx": _io_federation_actor_ctx,
|
|
"request-view-args": _io_request_view_args,
|
|
"cart-page-ctx": _io_cart_page_ctx,
|
|
"events-calendar-ctx": _io_events_calendar_ctx,
|
|
"events-day-ctx": _io_events_day_ctx,
|
|
"events-entry-ctx": _io_events_entry_ctx,
|
|
"events-slot-ctx": _io_events_slot_ctx,
|
|
"events-ticket-type-ctx": _io_events_ticket_type_ctx,
|
|
"market-header-ctx": _io_market_header_ctx,
|
|
"app-url": _io_app_url,
|
|
"asset-url": _io_asset_url,
|
|
"config": _io_config,
|
|
"jinja-global": _io_jinja_global,
|
|
"relations-from": _io_relations_from,
|
|
}
|
|
|
|
# Validate all I/O handlers are declared in boundary.sx
|
|
def _validate_io_handlers() -> None:
|
|
from .boundary import validate_io
|
|
for name in _IO_HANDLERS:
|
|
validate_io(name)
|
|
for name in IO_PRIMITIVES:
|
|
if name not in _IO_HANDLERS:
|
|
validate_io(name)
|
|
|
|
_validate_io_handlers()
|