diff --git a/account/app.py b/account/app.py
index 5b92a64..79e7600 100644
--- a/account/app.py
+++ b/account/app.py
@@ -1,6 +1,5 @@
from __future__ import annotations
import path_setup # noqa: F401 # adds shared/ to sys.path
-import sx.sx_components as sx_components # noqa: F401 # ensure Hypercorn --reload watches this file
from pathlib import Path
from quart import g, request
@@ -72,8 +71,9 @@ def create_app() -> "Quart":
app.jinja_loader,
])
- # Setup defpage routes
- import sx.sx_components # noqa: F811 — ensure components loaded
+ # Load .sx component files and setup defpage routes
+ from shared.sx.jinja_bridge import load_service_components
+ load_service_components(str(Path(__file__).resolve().parent), service_name="account")
from sxc.pages import setup_account_pages
setup_account_pages()
diff --git a/account/bp/account/routes.py b/account/bp/account/routes.py
index c4d2ce0..ca0f3bb 100644
--- a/account/bp/account/routes.py
+++ b/account/bp/account/routes.py
@@ -7,14 +7,13 @@ from __future__ import annotations
from quart import (
Blueprint,
- request,
g,
)
from sqlalchemy import select
from shared.models import UserNewsletter
from shared.infrastructure.fragments import fetch_fragments
-from shared.sx.helpers import sx_response
+from shared.sx.helpers import sx_response, render_to_sx
def register(url_prefix="/"):
@@ -55,7 +54,26 @@ def register(url_prefix="/"):
await g.s.flush()
- from sx.sx_components import render_newsletter_toggle
- return sx_response(await render_newsletter_toggle(un))
+ # Render toggle directly — no sx_components intermediary
+ from shared.browser.app.csrf import generate_csrf_token
+ from shared.infrastructure.urls import account_url
+
+ nid = un.newsletter_id
+ url_fn = getattr(g, "_account_url", None) or account_url
+ toggle_url = url_fn(f"/newsletter/{nid}/toggle/")
+ csrf = generate_csrf_token()
+ bg = "bg-emerald-500" if un.subscribed else "bg-stone-300"
+ translate = "translate-x-6" if un.subscribed else "translate-x-1"
+ checked = "true" if un.subscribed else "false"
+
+ return sx_response(await render_to_sx(
+ "account-newsletter-toggle",
+ id=f"nl-{nid}", url=toggle_url,
+ hdrs=f'{{"X-CSRFToken": "{csrf}"}}',
+ target=f"#nl-{nid}",
+ cls=f"relative inline-flex h-6 w-11 items-center rounded-full transition-colors focus:outline-none focus:ring-2 focus:ring-emerald-500 focus:ring-offset-2 {bg}",
+ checked=checked,
+ knob_cls=f"inline-block h-4 w-4 rounded-full bg-white shadow transform transition-transform {translate}",
+ ))
return account_bp
diff --git a/account/bp/auth/routes.py b/account/bp/auth/routes.py
index 4727051..a3cfdea 100644
--- a/account/bp/auth/routes.py
+++ b/account/bp/auth/routes.py
@@ -44,6 +44,17 @@ from .services import (
SESSION_USER_KEY = "uid"
ACCOUNT_SESSION_KEY = "account_sid"
+
+async def _render_auth_page(component: str, title: str, **kwargs) -> str:
+ """Render an auth page with root layout — replaces sx_components helpers."""
+ from shared.sx.helpers import render_to_sx, full_page_sx, root_header_sx
+ from shared.sx.page import get_template_context
+ ctx = await get_template_context()
+ hdr = await root_header_sx(ctx)
+ content = await render_to_sx(component, **{k: v for k, v in kwargs.items() if v})
+ return await full_page_sx(ctx, header_rows=hdr, content=content,
+ meta_html=f"
{title}")
+
ALLOWED_CLIENTS = {"blog", "market", "cart", "events", "federation", "orders", "test", "sx", "artdag", "artdag_l2"}
@@ -275,10 +286,7 @@ def register(url_prefix="/auth"):
redirect_url = pop_login_redirect_target()
return redirect(redirect_url)
- from shared.sx.page import get_template_context
- from sx.sx_components import render_login_page
- ctx = await get_template_context()
- return await render_login_page(ctx)
+ return await _render_auth_page("account-login-content", "Login \u2014 Rose Ash")
@rate_limit(
key_func=lambda: request.headers.get("X-Forwarded-For", request.remote_addr),
@@ -291,20 +299,20 @@ def register(url_prefix="/auth"):
is_valid, email = validate_email(email_input)
if not is_valid:
- from shared.sx.page import get_template_context
- from sx.sx_components import render_login_page
- ctx = await get_template_context(error="Please enter a valid email address.", email=email_input)
- return await render_login_page(ctx), 400
+ return await _render_auth_page(
+ "account-login-content", "Login \u2014 Rose Ash",
+ error="Please enter a valid email address.", email=email_input,
+ ), 400
# Per-email rate limit: 5 magic links per 15 minutes
from shared.infrastructure.rate_limit import _check_rate_limit
try:
allowed, _ = await _check_rate_limit(f"magic_email:{email}", 5, 900)
if not allowed:
- from shared.sx.page import get_template_context
- from sx.sx_components import render_check_email_page
- ctx = await get_template_context(email=email, email_error=None)
- return await render_check_email_page(ctx), 200
+ return await _render_auth_page(
+ "account-check-email-content", "Check your email \u2014 Rose Ash",
+ email=email,
+ ), 200
except Exception:
pass # Redis down — allow the request
@@ -324,10 +332,10 @@ def register(url_prefix="/auth"):
"Please try again in a moment."
)
- from shared.sx.page import get_template_context
- from sx.sx_components import render_check_email_page
- ctx = await get_template_context(email=email, email_error=email_error)
- return await render_check_email_page(ctx)
+ return await _render_auth_page(
+ "account-check-email-content", "Check your email \u2014 Rose Ash",
+ email=email, email_error=email_error,
+ )
@auth_bp.get("/magic//")
async def magic(token: str):
@@ -340,17 +348,17 @@ def register(url_prefix="/auth"):
user, error = await validate_magic_link(s, token)
if error:
- from shared.sx.page import get_template_context
- from sx.sx_components import render_login_page
- ctx = await get_template_context(error=error)
- return await render_login_page(ctx), 400
+ return await _render_auth_page(
+ "account-login-content", "Login \u2014 Rose Ash",
+ error=error,
+ ), 400
user_id = user.id
except Exception:
- from shared.sx.page import get_template_context
- from sx.sx_components import render_login_page
- ctx = await get_template_context(error="Could not sign you in right now. Please try again.")
- return await render_login_page(ctx), 502
+ return await _render_auth_page(
+ "account-login-content", "Login \u2014 Rose Ash",
+ error="Could not sign you in right now. Please try again.",
+ ), 502
assert user_id is not None
@@ -679,11 +687,11 @@ def register(url_prefix="/auth"):
@auth_bp.get("/device/")
async def device_form():
"""Browser form where user enters the code displayed in terminal."""
- from shared.sx.page import get_template_context
- from sx.sx_components import render_device_page
code = request.args.get("code", "")
- ctx = await get_template_context(code=code)
- return await render_device_page(ctx)
+ return await _render_auth_page(
+ "account-device-content", "Authorize Device \u2014 Rose Ash",
+ code=code,
+ )
@auth_bp.post("/device")
@auth_bp.post("/device/")
@@ -693,20 +701,20 @@ def register(url_prefix="/auth"):
user_code = (form.get("code") or "").strip().replace("-", "").upper()
if not user_code or len(user_code) != 8:
- from shared.sx.page import get_template_context
- from sx.sx_components import render_device_page
- ctx = await get_template_context(error="Please enter a valid 8-character code.", code=form.get("code", ""))
- return await render_device_page(ctx), 400
+ return await _render_auth_page(
+ "account-device-content", "Authorize Device \u2014 Rose Ash",
+ error="Please enter a valid 8-character code.", code=form.get("code", ""),
+ ), 400
from shared.infrastructure.auth_redis import get_auth_redis
r = await get_auth_redis()
device_code = await r.get(f"devflow_uc:{user_code}")
if not device_code:
- from shared.sx.page import get_template_context
- from sx.sx_components import render_device_page
- ctx = await get_template_context(error="Code not found or expired. Please try again.", code=form.get("code", ""))
- return await render_device_page(ctx), 400
+ return await _render_auth_page(
+ "account-device-content", "Authorize Device \u2014 Rose Ash",
+ error="Code not found or expired. Please try again.", code=form.get("code", ""),
+ ), 400
if isinstance(device_code, bytes):
device_code = device_code.decode()
@@ -720,23 +728,19 @@ def register(url_prefix="/auth"):
# Logged in — approve immediately
ok = await _approve_device(device_code, g.user)
if not ok:
- from shared.sx.page import get_template_context
- from sx.sx_components import render_device_page
- ctx = await get_template_context(error="Code expired or already used.")
- return await render_device_page(ctx), 400
+ return await _render_auth_page(
+ "account-device-content", "Authorize Device \u2014 Rose Ash",
+ error="Code expired or already used.",
+ ), 400
- from shared.sx.page import get_template_context
- from sx.sx_components import render_device_approved_page
- ctx = await get_template_context()
- return await render_device_approved_page(ctx)
+ return await _render_auth_page(
+ "account-device-approved", "Device Authorized \u2014 Rose Ash",
+ )
@auth_bp.get("/device/complete")
@auth_bp.get("/device/complete/")
async def device_complete():
"""Post-login redirect — completes approval after magic link auth."""
- from shared.sx.page import get_template_context
- from sx.sx_components import render_device_page, render_device_approved_page
-
device_code = request.args.get("code", "")
if not device_code:
@@ -748,12 +752,13 @@ def register(url_prefix="/auth"):
ok = await _approve_device(device_code, g.user)
if not ok:
- ctx = await get_template_context(
+ return await _render_auth_page(
+ "account-device-content", "Authorize Device \u2014 Rose Ash",
error="Code expired or already used. Please start the login process again in your terminal.",
- )
- return await render_device_page(ctx), 400
+ ), 400
- ctx = await get_template_context()
- return await render_device_approved_page(ctx)
+ return await _render_auth_page(
+ "account-device-approved", "Device Authorized \u2014 Rose Ash",
+ )
return auth_bp
diff --git a/account/sx/sx_components.py b/account/sx/sx_components.py
deleted file mode 100644
index 9073111..0000000
--- a/account/sx/sx_components.py
+++ /dev/null
@@ -1,107 +0,0 @@
-"""
-Account service s-expression page components.
-
-Renders login, device auth, and check-email pages. Dashboard and newsletters
-are now fully handled by .sx defcomps called from defpage expressions.
-"""
-from __future__ import annotations
-
-import os
-from typing import Any
-
-from shared.sx.jinja_bridge import load_service_components
-from shared.sx.helpers import (
- render_to_sx,
- root_header_sx, full_page_sx,
-)
-
-# Load account-specific .sx components + handlers at import time
-load_service_components(os.path.dirname(os.path.dirname(__file__)),
- service_name="account")
-
-
-# ---------------------------------------------------------------------------
-# Public API: Auth pages (login, device, check_email)
-# ---------------------------------------------------------------------------
-
-async def render_login_page(ctx: dict) -> str:
- """Full page: login form."""
- error = ctx.get("error", "")
- email = ctx.get("email", "")
- hdr = await root_header_sx(ctx)
- content = await render_to_sx("account-login-content",
- error=error or None, email=email)
- return await full_page_sx(ctx, header_rows=hdr,
- content=content,
- meta_html='Login \u2014 Rose Ash')
-
-
-async def render_device_page(ctx: dict) -> str:
- """Full page: device authorization form."""
- error = ctx.get("error", "")
- code = ctx.get("code", "")
- hdr = await root_header_sx(ctx)
- content = await render_to_sx("account-device-content",
- error=error or None, code=code)
- return await full_page_sx(ctx, header_rows=hdr,
- content=content,
- meta_html='Authorize Device \u2014 Rose Ash')
-
-
-async def render_device_approved_page(ctx: dict) -> str:
- """Full page: device approved."""
- hdr = await root_header_sx(ctx)
- content = await render_to_sx("account-device-approved")
- return await full_page_sx(ctx, header_rows=hdr,
- content=content,
- meta_html='Device Authorized \u2014 Rose Ash')
-
-
-async def render_check_email_page(ctx: dict) -> str:
- """Full page: check email after magic link sent."""
- email = ctx.get("email", "")
- email_error = ctx.get("email_error")
- hdr = await root_header_sx(ctx)
- content = await render_to_sx("account-check-email-content",
- email=email, email_error=email_error)
- return await full_page_sx(ctx, header_rows=hdr,
- content=content,
- meta_html='Check your email \u2014 Rose Ash')
-
-
-# ---------------------------------------------------------------------------
-# Public API: Fragment renderers for POST handlers
-# ---------------------------------------------------------------------------
-
-async def render_newsletter_toggle(un) -> str:
- """Render a newsletter toggle switch for POST response."""
- from shared.browser.app.csrf import generate_csrf_token
-
- nid = un.newsletter_id
- from quart import g
- account_url_fn = getattr(g, "_account_url", None)
- if account_url_fn is None:
- from shared.infrastructure.urls import account_url
- account_url_fn = account_url
-
- toggle_url = account_url_fn(f"/newsletter/{nid}/toggle/")
- csrf = generate_csrf_token()
-
- if un.subscribed:
- bg = "bg-emerald-500"
- translate = "translate-x-6"
- checked = "true"
- else:
- bg = "bg-stone-300"
- translate = "translate-x-1"
- checked = "false"
-
- return await render_to_sx(
- "account-newsletter-toggle",
- id=f"nl-{nid}", url=toggle_url,
- hdrs=f'{{"X-CSRFToken": "{csrf}"}}',
- target=f"#nl-{nid}",
- cls=f"relative inline-flex h-6 w-11 items-center rounded-full transition-colors focus:outline-none focus:ring-2 focus:ring-emerald-500 focus:ring-offset-2 {bg}",
- checked=checked,
- knob_cls=f"inline-block h-4 w-4 rounded-full bg-white shadow transform transition-transform {translate}",
- )
diff --git a/account/sxc/pages/__init__.py b/account/sxc/pages/__init__.py
index 3624fb7..bd960a0 100644
--- a/account/sxc/pages/__init__.py
+++ b/account/sxc/pages/__init__.py
@@ -54,6 +54,7 @@ async def _account_oob(ctx: dict, **kw: Any) -> str:
async def _account_mobile(ctx: dict, **kw: Any) -> str:
from shared.sx.helpers import mobile_menu_sx, mobile_root_nav_sx, render_to_sx
from shared.sx.parser import SxExpr
+
ctx = _inject_account_nav(ctx)
nav_items = await render_to_sx("auth-nav-items",
account_url=_call_url(ctx, "account_url", ""),
@@ -97,18 +98,16 @@ def _register_account_helpers() -> None:
from shared.sx.pages import register_page_helpers
register_page_helpers("account", {
- "newsletters-content": _h_newsletters_content,
- "fragment-content": _h_fragment_content,
+ "newsletters-data": _h_newsletters_data,
})
-async def _h_newsletters_content(**kw):
- """Fetch newsletter data, return assembled defcomp call."""
+async def _h_newsletters_data(**kw):
+ """Fetch newsletter data — returns dict merged into defpage env."""
from quart import g
from sqlalchemy import select
from shared.models import UserNewsletter
from shared.models.ghost_membership_entities import GhostNewsletter
- from shared.sx.helpers import render_to_sx
result = await g.s.execute(
select(GhostNewsletter).order_by(GhostNewsletter.name)
@@ -135,31 +134,6 @@ async def _h_newsletters_content(**kw):
if account_url is None:
from shared.infrastructure.urls import account_url as _account_url
account_url = _account_url
- # Call account_url to get the base URL string
account_url_str = account_url("") if callable(account_url) else str(account_url or "")
- return await render_to_sx("account-newsletters-content",
- newsletter_list=newsletter_list,
- account_url=account_url_str)
-
-
-async def _h_fragment_content(slug=None, **kw):
- from quart import g, abort
- from shared.infrastructure.fragments import fetch_fragment
-
- if not slug or not g.get("user"):
- return ""
- fragment_html = await fetch_fragment(
- "events", "account-page",
- params={"slug": slug, "user_id": str(g.user.id)},
- )
- if not fragment_html:
- abort(404)
- from shared.sx.parser import SxExpr
- if isinstance(fragment_html, SxExpr):
- return fragment_html.source
- s = str(fragment_html) if fragment_html else ""
- if not s:
- return ""
- from shared.sx.helpers import render_to_sx
- return await render_to_sx("rich-text", html=s)
+ return {"newsletter-list": newsletter_list, "account-url": account_url_str}
diff --git a/account/sxc/pages/account.sx b/account/sxc/pages/account.sx
index 8da12d1..f85e807 100644
--- a/account/sxc/pages/account.sx
+++ b/account/sxc/pages/account.sx
@@ -18,7 +18,10 @@
:path "/newsletters/"
:auth :login
:layout :account
- :content (newsletters-content))
+ :data (newsletters-data)
+ :content (~account-newsletters-content
+ :newsletter-list newsletter-list
+ :account-url account-url))
;; ---------------------------------------------------------------------------
;; Fragment pages (tickets, bookings, etc. from events service)
@@ -28,4 +31,10 @@
:path "//"
:auth :login
:layout :account
- :content (fragment-content slug))
+ :content (let* ((user (current-user))
+ (result (frag "events" "account-page"
+ :slug slug
+ :user-id (str (get user "id")))))
+ (if (or (nil? result) (empty? result))
+ (abort 404)
+ result)))
diff --git a/events/sx/sx_components.py b/events/sx/sx_components.py
index 338e6db..d4fb6b1 100644
--- a/events/sx/sx_components.py
+++ b/events/sx/sx_components.py
@@ -198,7 +198,7 @@ async def _calendar_nav_sx(ctx: dict) -> str:
admin_href = url_for("defpage_calendar_admin", calendar_slug=cal_slug)
parts.append(await render_to_sx("nav-link", href=admin_href, icon="fa fa-cog",
select_colours=select_colours))
- return "".join(parts)
+ return "(<> " + " ".join(parts) + ")" if parts else ""
# ---------------------------------------------------------------------------
diff --git a/market/sx/sx_components.py b/market/sx/sx_components.py
index 7519e66..931045b 100644
--- a/market/sx/sx_components.py
+++ b/market/sx/sx_components.py
@@ -111,7 +111,7 @@ async def _market_header_sx(ctx: dict, *, oob: bool = False) -> str:
sub_div=SxExpr(sub_div) if sub_div else None,
)
- link_href = url_for("market.browse.defpage_market_home")
+ link_href = url_for("defpage_market_home")
# Build desktop nav from categories
categories = ctx.get("categories", {})
diff --git a/shared/sx/async_eval.py b/shared/sx/async_eval.py
index ef8a5d8..b89eba9 100644
--- a/shared/sx/async_eval.py
+++ b/shared/sx/async_eval.py
@@ -910,6 +910,42 @@ async def async_eval_to_sx(
return serialize(result)
+async def async_eval_slot_to_sx(
+ expr: Any,
+ env: dict[str, Any],
+ ctx: RequestContext | None = None,
+) -> str:
+ """Like async_eval_to_sx but expands component calls.
+
+ Used by defpage slot evaluation where the content expression is
+ typically a component call like ``(~dashboard-content)``. Normal
+ ``async_eval_to_sx`` serializes component calls without expanding;
+ this variant expands one level so IO primitives in the body execute,
+ then serializes the result as SX wire format.
+ """
+ if ctx is None:
+ ctx = RequestContext()
+ # If expr is a component call, expand it through _aser
+ if isinstance(expr, list) and expr:
+ head = expr[0]
+ if isinstance(head, Symbol) and head.name.startswith("~"):
+ comp = env.get(head.name)
+ if isinstance(comp, Component):
+ result = await _aser_component(comp, expr[1:], env, ctx)
+ if isinstance(result, SxExpr):
+ return result.source
+ if result is None or result is NIL:
+ return ""
+ return serialize(result)
+ # Fall back to normal async_eval_to_sx
+ result = await _aser(expr, env, ctx)
+ if isinstance(result, SxExpr):
+ return result.source
+ if result is None or result is NIL:
+ return ""
+ return serialize(result)
+
+
async def _aser(expr: Any, env: dict[str, Any], ctx: RequestContext) -> Any:
"""Evaluate *expr*, producing SxExpr for rendering forms, raw values
for everything else."""
@@ -1022,6 +1058,33 @@ async def _aser_fragment(children: list, env: dict, ctx: RequestContext) -> SxEx
return SxExpr("(<> " + " ".join(parts) + ")")
+async def _aser_component(
+ comp: Component, args: list, env: dict, ctx: RequestContext,
+) -> Any:
+ """Expand a component body through _aser — produces SX, not HTML."""
+ kwargs: dict[str, Any] = {}
+ children: list[Any] = []
+ i = 0
+ while i < len(args):
+ arg = args[i]
+ if isinstance(arg, Keyword) and i + 1 < len(args):
+ kwargs[arg.name] = await async_eval(args[i + 1], env, ctx)
+ i += 2
+ else:
+ children.append(arg)
+ i += 1
+ local = dict(comp.closure)
+ local.update(env)
+ for p in comp.params:
+ local[p] = kwargs.get(p, NIL)
+ if comp.has_children:
+ child_parts = []
+ for c in children:
+ child_parts.append(serialize(await _aser(c, env, ctx)))
+ local["children"] = SxExpr("(<> " + " ".join(child_parts) + ")")
+ return await _aser(comp.body, local, ctx)
+
+
async def _aser_call(
name: str, args: list, env: dict, ctx: RequestContext,
) -> SxExpr:
diff --git a/shared/sx/pages.py b/shared/sx/pages.py
index 7881b6c..a1cba1d 100644
--- a/shared/sx/pages.py
+++ b/shared/sx/pages.py
@@ -132,31 +132,14 @@ def load_page_dir(directory: str, service_name: str) -> list[PageDef]:
# Page execution
# ---------------------------------------------------------------------------
-async def _eval_slot(expr: Any, env: dict, ctx: Any,
- async_eval_fn: Any, async_eval_to_sx_fn: Any) -> str:
+async def _eval_slot(expr: Any, env: dict, ctx: 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.
+ Expands component calls (so IO in the body executes) but serializes
+ the result as SX wire format, not HTML.
"""
- 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)
+ from .async_eval import async_eval_slot_to_sx
+ return await async_eval_slot_to_sx(expr, env, ctx)
async def execute_page(
@@ -174,7 +157,7 @@ async def execute_page(
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 .async_eval import async_eval
from .page import get_template_context
from .helpers import full_page_sx, oob_page_sx, sx_response
from .layouts import get_layout
@@ -204,20 +187,20 @@ async def execute_page(
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)
+ content_sx = await _eval_slot(page_def.content_expr, env, ctx)
# 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)
+ filter_sx = await _eval_slot(page_def.filter_expr, env, ctx)
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)
+ aside_sx = await _eval_slot(page_def.aside_expr, env, ctx)
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)
+ menu_sx = await _eval_slot(page_def.menu_expr, env, ctx)
# Resolve layout → header rows + mobile menu fallback
tctx = await get_template_context()