""" 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 IO handlers (name → coroutine) # --------------------------------------------------------------------------- _IO_HANDLERS: dict[str, Any] = {} def register_io_handler(name: str): """Decorator that registers an async function as an IO handler.""" def decorator(fn): _IO_HANDLERS[name] = fn return fn return decorator # --------------------------------------------------------------------------- # Request context (set per-request by the resolver) # --------------------------------------------------------------------------- _handler_service: contextvars.ContextVar[Any] = contextvars.ContextVar( "_handler_service", default=None ) _response_meta: contextvars.ContextVar[dict | None] = contextvars.ContextVar( "_response_meta", default=None ) # Ephemeral per-process state — resets on restart. For demos/testing only. _ephemeral_state: dict[str, Any] = {} 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) def reset_response_meta() -> None: """Reset response meta for a new request.""" _response_meta.set(None) def get_response_meta() -> dict | None: """Get response meta (headers/status) set by handler IO primitives.""" return _response_meta.get(None) class RequestContext: """Per-request context provided to I/O primitives.""" __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.""" handler = _IO_HANDLERS.get(name) if handler is None: raise RuntimeError(f"Unknown I/O primitive: {name}") return await handler(args, kwargs, ctx) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- 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} def _dto_to_dict(obj: Any) -> dict[str, Any]: """Convert a DTO/dataclass/namedtuple to a plain dict.""" 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} 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.""" if result is None: from .types import NIL return NIL if isinstance(result, (int, float, str, bool)): return result 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): 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 # --------------------------------------------------------------------------- # Generic IO handlers # --------------------------------------------------------------------------- @register_io_handler("frag") 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) @register_io_handler("query") 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) @register_io_handler("action") 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) @register_io_handler("current-user") 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 @register_io_handler("htmx-request?") 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 @register_io_handler("service") async def _io_service( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> Any: """``(service "svc-name" "method-name" :key val ...)`` → call domain service.""" if not args: raise ValueError("service requires at least a method name") if len(args) >= 2: 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: 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}") 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) @register_io_handler("request-arg") 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) @register_io_handler("request-path") 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 @register_io_handler("nav-tree") 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] @register_io_handler("get-children") 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] @register_io_handler("g") async def _io_g( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> Any: """``(g "key")`` → getattr(g, key, None).""" from quart import g key = str(args[0]).replace("-", "_") if args else "" 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("request-args-all") async def _io_request_args_all( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> dict: """``(request-args-all)`` → all query params as dict.""" from quart import request return dict(request.args) @register_io_handler("request-form-all") async def _io_request_form_all( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> dict: """``(request-form-all)`` → all form fields as dict.""" from quart import request form = await request.form return dict(form) @register_io_handler("request-form-list") async def _io_request_form_list( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> list: """``(request-form-list "field")`` → all values for a multi-value form field.""" if not args: raise ValueError("request-form-list requires a field name") from quart import request form = await request.form return form.getlist(str(args[0])) @register_io_handler("request-headers-all") async def _io_request_headers_all( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> dict: """``(request-headers-all)`` → all headers as dict (lowercase keys).""" from quart import request return {k.lower(): v for k, v in request.headers} @register_io_handler("request-file-name") async def _io_request_file_name( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> Any: """``(request-file-name "field")`` → filename or nil.""" if not args: raise ValueError("request-file-name requires a field name") from quart import request from .types import NIL files = await request.files f = files.get(str(args[0])) return f.filename if f else NIL @register_io_handler("set-response-header") async def _io_set_response_header( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> Any: """``(set-response-header "Name" "value")`` → set on response after handler.""" if len(args) < 2: raise ValueError("set-response-header requires name and value") from .types import NIL meta = _response_meta.get(None) if meta is None: meta = {"headers": {}, "status": None} _response_meta.set(meta) meta["headers"][str(args[0])] = str(args[1]) return NIL @register_io_handler("set-response-status") async def _io_set_response_status( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> Any: """``(set-response-status 503)`` → set status code on response.""" if not args: raise ValueError("set-response-status requires a status code") from .types import NIL meta = _response_meta.get(None) if meta is None: meta = {"headers": {}, "status": None} _response_meta.set(meta) meta["status"] = int(args[0]) return NIL @register_io_handler("state-get") async def _io_state_get( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> Any: """``(state-get "key" default?)`` → read from ephemeral state.""" if not args: raise ValueError("state-get requires a key") from .types import NIL key = str(args[0]) default = args[1] if len(args) > 1 else NIL return _ephemeral_state.get(key, default) @register_io_handler("state-set!") async def _io_state_set( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> Any: """``(state-set! "key" value)`` → write to ephemeral state.""" if len(args) < 2: raise ValueError("state-set! requires key and value") from .types import NIL _ephemeral_state[str(args[0])] = args[1] return NIL @register_io_handler("csrf-token") 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 "" @register_io_handler("abort") async def _io_abort( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> Any: """``(abort 403 "message")`` — raise HTTP error from SX.""" 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) @register_io_handler("url-for") async def _io_url_for( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> str: """``(url-for "endpoint" :key val ...)`` → url_for(endpoint, **kwargs).""" 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()} for k, v in clean.items(): if isinstance(v, str) and v.isdigit(): clean[k] = int(v) return url_for(endpoint, **clean) @register_io_handler("route-prefix") 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() @register_io_handler("request-view-args") 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) @register_io_handler("app-url") 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) @register_io_handler("asset-url") async def _io_asset_url( args: list[Any], kwargs: dict[str, Any], ctx: RequestContext ) -> str: """``(asset-url "/img/logo.png")`` → versioned static URL.""" from quart import current_app asset_url = current_app.jinja_env.globals.get("asset_url") if asset_url is None: raise RuntimeError("asset_url Jinja global not registered") path = str(args[0]) if args else "" return asset_url(path) @register_io_handler("config") 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])) @register_io_handler("jinja-global") 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) @register_io_handler("relations-from") 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])) ] # --------------------------------------------------------------------------- # Import context handlers (registers into _IO_HANDLERS via decorator) # --------------------------------------------------------------------------- from . import primitives_ctx # noqa: E402, F401 # --------------------------------------------------------------------------- # Auto-derive IO_PRIMITIVES from registered handlers # --------------------------------------------------------------------------- IO_PRIMITIVES: frozenset[str] = frozenset(_IO_HANDLERS.keys()) # --------------------------------------------------------------------------- # Sync IO bridge primitives # # These are declared in boundary.sx (I/O tier), NOT primitives.sx. # They must be evaluator-visible because they're called inline in .sx # code (inside let, filter, etc.) where the async IO interceptor can't # reach them — particularly in sx_ref.py which only intercepts IO at # the top level. # --------------------------------------------------------------------------- from .primitives import _PRIMITIVES # noqa: E402 def _bridge_app_url(service, *path_parts): from shared.infrastructure.urls import app_url path = str(path_parts[0]) if path_parts else "/" return app_url(str(service), path) def _bridge_asset_url(*path_parts): from quart import current_app asset_url = current_app.jinja_env.globals.get("asset_url") if asset_url is None: raise RuntimeError("asset_url Jinja global not registered") path = str(path_parts[0]) if path_parts else "" return asset_url(path) def _bridge_config(key): from shared.config import config cfg = config() return cfg.get(str(key)) def _bridge_jinja_global(key, *default): from quart import current_app d = default[0] if default else None return current_app.jinja_env.globals.get(str(key), d) def _bridge_relations_from(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(entity_type)) ] _PRIMITIVES["app-url"] = _bridge_app_url _PRIMITIVES["asset-url"] = _bridge_asset_url _PRIMITIVES["config"] = _bridge_config _PRIMITIVES["jinja-global"] = _bridge_jinja_global _PRIMITIVES["relations-from"] = _bridge_relations_from # --------------------------------------------------------------------------- # Validate all IO handlers against boundary.sx # --------------------------------------------------------------------------- def _validate_io_handlers() -> None: from .boundary import validate_io for name in _IO_HANDLERS: validate_io(name) _validate_io_handlers()