Files
mono/shared/sx/primitives_io.py
giles e8bc228c7f Rebrand sexp → sx across web platform (173 files)
Rename all sexp directories, files, identifiers, and references to sx.
artdag/ excluded (separate media processing DSL).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-01 11:06:57 +00:00

154 lines
4.9 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
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?",
})
# ---------------------------------------------------------------------------
# Request context (set per-request by the resolver)
# ---------------------------------------------------------------------------
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
# ---------------------------------------------------------------------------
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 = {k: v for k, v in kwargs.items() if v is not None}
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 = {k: v for k, v in kwargs.items() if v is not None}
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 = {k: v for k, v in kwargs.items() if v is not None}
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
# ---------------------------------------------------------------------------
# Handler registry
# ---------------------------------------------------------------------------
_IO_HANDLERS: dict[str, Any] = {
"frag": _io_frag,
"query": _io_query,
"action": _io_action,
"current-user": _io_current_user,
"htmx-request?": _io_htmx_request,
}