Files
rose-ash/shared/sx/types.py
giles 1f36987f77
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 2m5s
Replace inter-service _handlers dicts with declarative sx defquery/defaction
The inter-service data layer (fetch_data/call_action) was the least
structured part of the codebase — Python _handlers dicts with ad-hoc
param extraction scattered across 16 route files. This replaces them
with declarative .sx query/action definitions that make the entire
inter-service protocol self-describing and greppable.

Infrastructure:
- defquery/defaction special forms in the sx evaluator
- Query/action registry with load, lookup, and schema introspection
- Query executor using async_eval with I/O primitives
- Blueprint factories (create_data_blueprint/create_action_blueprint)
  with sx-first dispatch and Python fallback
- /internal/schema endpoint on every service
- parse-datetime and split-ids primitives for type coercion

Service extractions:
- LikesService (toggle, is_liked, liked_slugs, liked_ids)
- PageConfigService (ensure, get_by_container, get_by_id, get_batch, update)
- RelationsService (wraps module-level functions)
- AccountDataService (user_by_email, newsletters)
- CartItemsService, MarketDataService (raw SQLAlchemy lookups)

50 of 54 handlers converted to sx, 4 Python fallbacks remain
(ghost-sync/push-member, clear-cart-for-order, create-order).
Net: -1,383 lines Python, +251 lines modified.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-04 08:13:50 +00:00

287 lines
9.1 KiB
Python

"""
Core types for the s-expression language.
Symbol — unquoted identifier (e.g. div, ~card, map)
Keyword — colon-prefixed key (e.g. :class, :id)
Lambda — callable closure created by (lambda ...) or (fn ...)
Nil — singleton null value
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
# ---------------------------------------------------------------------------
# Nil
# ---------------------------------------------------------------------------
class _Nil:
"""Singleton nil value — falsy, serialises as 'nil'."""
_instance: _Nil | None = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __bool__(self):
return False
def __repr__(self):
return "nil"
def __eq__(self, other):
return other is None or isinstance(other, _Nil)
def __hash__(self):
return hash(None)
NIL = _Nil()
# ---------------------------------------------------------------------------
# Symbol
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Symbol:
"""An unquoted symbol/identifier."""
name: str
def __repr__(self):
return f"Symbol({self.name!r})"
def __eq__(self, other):
if isinstance(other, Symbol):
return self.name == other.name
if isinstance(other, str):
return self.name == other
return False
def __hash__(self):
return hash(self.name)
@property
def is_component(self) -> bool:
"""True if this symbol names a component (~prefix)."""
return self.name.startswith("~")
# ---------------------------------------------------------------------------
# Keyword
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Keyword:
"""A keyword starting with colon (e.g. :class, :id)."""
name: str
def __repr__(self):
return f"Keyword({self.name!r})"
def __eq__(self, other):
if isinstance(other, Keyword):
return self.name == other.name
return False
def __hash__(self):
return hash((":", self.name))
# ---------------------------------------------------------------------------
# Lambda
# ---------------------------------------------------------------------------
@dataclass
class Lambda:
"""A callable closure.
Created by ``(lambda (x) body)`` or ``(fn (x) body)``.
Captures the defining environment so free variables resolve correctly.
"""
params: list[str]
body: Any
closure: dict[str, Any] = field(default_factory=dict)
name: str | None = None # optional, set by (define name (fn ...))
def __repr__(self):
tag = self.name or "lambda"
return f"<{tag}({', '.join(self.params)})>"
def __call__(self, *args: Any, evaluator: Any = None, caller_env: dict | None = None) -> Any:
"""Invoke the lambda. Requires *evaluator* — the evaluate() function."""
if evaluator is None:
raise RuntimeError("Lambda requires evaluator to be called")
if len(args) != len(self.params):
raise RuntimeError(
f"{self!r} expects {len(self.params)} args, got {len(args)}"
)
local = dict(self.closure)
if caller_env:
local.update(caller_env)
for p, v in zip(self.params, args):
local[p] = v
return evaluator(self.body, local)
# ---------------------------------------------------------------------------
# Macro
# ---------------------------------------------------------------------------
@dataclass
class Macro:
"""A macro — an AST-transforming function.
Created by ``(defmacro name (params... &rest rest) body)``.
Receives unevaluated arguments, evaluates its body to produce a new
s-expression, which is then evaluated in the caller's environment.
"""
params: list[str]
rest_param: str | None # &rest parameter name
body: Any # unevaluated — returns an s-expression to eval
closure: dict[str, Any] = field(default_factory=dict)
name: str | None = None
def __repr__(self):
tag = self.name or "macro"
return f"<{tag}({', '.join(self.params)})>"
# ---------------------------------------------------------------------------
# Component
# ---------------------------------------------------------------------------
@dataclass
class Component:
"""A reusable UI component defined via ``(defcomp ~name (&key ...) body)``.
Components are like lambdas but accept keyword arguments and support
a ``children`` rest parameter.
"""
name: str
params: list[str] # keyword parameter names (without &key prefix)
has_children: bool # True if &rest children declared
body: Any # unevaluated s-expression body
closure: dict[str, Any] = field(default_factory=dict)
css_classes: set[str] = field(default_factory=set) # pre-scanned :class values
def __repr__(self):
return f"<Component ~{self.name}({', '.join(self.params)})>"
# ---------------------------------------------------------------------------
# HandlerDef
# ---------------------------------------------------------------------------
@dataclass
class HandlerDef:
"""A declarative fragment handler defined in an .sx file.
Created by ``(defhandler name (&key param...) body)``.
The body is evaluated in a sandboxed environment with only
s-expression primitives available.
"""
name: str
params: list[str] # keyword parameter names
body: Any # unevaluated s-expression body
closure: dict[str, Any] = field(default_factory=dict)
def __repr__(self):
return f"<handler:{self.name}({', '.join(self.params)})>"
# ---------------------------------------------------------------------------
# RelationDef
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class RelationDef:
"""A declared relation between two entity types.
Created by ``(defrelation :name ...)`` s-expressions.
"""
name: str # "page->market"
from_type: str # "page"
to_type: str # "market"
cardinality: str # "one-to-one" | "one-to-many" | "many-to-many"
inverse: str | None # "market->page"
nav: str # "submenu" | "tab" | "badge" | "inline" | "hidden"
nav_icon: str | None # "fa fa-shopping-bag"
nav_label: str | None # "markets"
# ---------------------------------------------------------------------------
# PageDef
# ---------------------------------------------------------------------------
@dataclass
class PageDef:
"""A declarative GET page defined in an .sx file.
Created by ``(defpage name :path "/..." :auth :public :content expr)``.
Slots are stored as unevaluated AST and resolved at request time.
"""
name: str
path: str
auth: str | list # "public", "login", "admin", or ["rights", ...]
layout: Any # layout name/config (unevaluated)
cache: dict | None
data_expr: Any # unevaluated AST
content_expr: Any # unevaluated AST
filter_expr: Any
aside_expr: Any
menu_expr: Any
closure: dict[str, Any] = field(default_factory=dict)
def __repr__(self):
return f"<page:{self.name} path={self.path!r}>"
# ---------------------------------------------------------------------------
# QueryDef / ActionDef
# ---------------------------------------------------------------------------
@dataclass
class QueryDef:
"""A declarative data query defined in an .sx file.
Created by ``(defquery name (&key param...) "docstring" body)``.
The body is evaluated with async I/O primitives to produce JSON data.
"""
name: str
params: list[str] # keyword parameter names
doc: str # docstring
body: Any # unevaluated s-expression body
closure: dict[str, Any] = field(default_factory=dict)
def __repr__(self):
return f"<query:{self.name}({', '.join(self.params)})>"
@dataclass
class ActionDef:
"""A declarative action defined in an .sx file.
Created by ``(defaction name (&key param...) "docstring" body)``.
The body is evaluated with async I/O primitives to produce JSON data.
"""
name: str
params: list[str] # keyword parameter names
doc: str # docstring
body: Any # unevaluated s-expression body
closure: dict[str, Any] = field(default_factory=dict)
def __repr__(self):
return f"<action:{self.name}({', '.join(self.params)})>"
# ---------------------------------------------------------------------------
# Type alias
# ---------------------------------------------------------------------------
# An s-expression value after evaluation
SExp = int | float | str | bool | Symbol | Keyword | Lambda | Macro | Component | HandlerDef | RelationDef | PageDef | QueryDef | ActionDef | list | dict | _Nil | None