From dacb61b0aef5bfef961882e12a73704c46d2e077 Mon Sep 17 00:00:00 2001 From: giles Date: Wed, 4 Mar 2026 01:22:33 +0000 Subject: [PATCH] =?UTF-8?q?Delete=20orders=20+=20federation=20sx=5Fcompone?= =?UTF-8?q?nts.py=20=E2=80=94=20rendering=20inlined=20to=20routes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2 (Orders): - Checkout error/return renders moved directly into route handlers - Removed orphaned test_sx_helpers.py Phase 3 (Federation): - Auth pages use _render_social_auth_page() helper in routes - Choose-username render inlined into identity routes - Timeline/search/follow/interaction renders inlined into social routes using serializers imported from sxc.pages - Added _social_page() to sxc/pages/__init__.py for shared use - Home page renders inline in app.py Co-Authored-By: Claude Opus 4.6 --- federation/app.py | 10 +- federation/bp/auth/routes.py | 47 ++--- federation/bp/identity/routes.py | 39 ++-- federation/bp/social/routes.py | 145 +++++++++++---- federation/sx/sx_components.py | 293 ------------------------------- federation/sxc/pages/__init__.py | 17 ++ orders/app.py | 9 +- orders/bp/checkout/routes.py | 117 +++++++++++- orders/bp/order/routes.py | 17 +- orders/sx/sx_components.py | 189 -------------------- orders/tests/test_sx_helpers.py | 32 ---- 11 files changed, 326 insertions(+), 589 deletions(-) delete mode 100644 federation/sx/sx_components.py delete mode 100644 orders/sx/sx_components.py delete mode 100644 orders/tests/test_sx_helpers.py diff --git a/federation/app.py b/federation/app.py index 5b86d3d..f652ae6 100644 --- a/federation/app.py +++ b/federation/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 @@ -83,7 +82,9 @@ def create_app() -> "Quart": app.jinja_loader, ]) - # --- defpage setup --- + # 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="federation") from sxc.pages import setup_federation_pages setup_federation_pages() @@ -106,10 +107,11 @@ def create_app() -> "Quart": async def home(): from quart import make_response from shared.sx.page import get_template_context - from sx.sx_components import render_federation_home + from shared.sx.helpers import root_header_sx, full_page_sx ctx = await get_template_context() - html = await render_federation_home(ctx) + hdr = await root_header_sx(ctx) + html = await full_page_sx(ctx, header_rows=hdr) return await make_response(html) return app diff --git a/federation/bp/auth/routes.py b/federation/bp/auth/routes.py index f634d30..16139fd 100644 --- a/federation/bp/auth/routes.py +++ b/federation/bp/auth/routes.py @@ -42,6 +42,16 @@ SESSION_USER_KEY = "uid" ALLOWED_CLIENTS = {"blog", "market", "cart", "events", "account"} +async def _render_social_auth_page(component: str, title: str, **kwargs) -> str: + """Render an auth page with social layout — replaces sx_components helpers.""" + from shared.sx.helpers import render_to_sx + from shared.sx.page import get_template_context + from sxc.pages import _social_page + ctx = await get_template_context() + content = await render_to_sx(component, **{k: v for k, v in kwargs.items() if v}) + return await _social_page(ctx, None, content=content, title=title) + + def register(url_prefix="/auth"): auth_bp = Blueprint("auth", __name__, url_prefix=url_prefix) @@ -99,10 +109,7 @@ def register(url_prefix="/auth"): # If there's a pending redirect (e.g. OAuth authorize), follow it 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_social_auth_page("account-login-content", "Login \u2014 Rose Ash") @auth_bp.post("/start/") async def start_login(): @@ -111,10 +118,10 @@ 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_social_auth_page( + "account-login-content", "Login \u2014 Rose Ash", + error="Please enter a valid email address.", email=email_input, + ), 400 user = await find_or_create_user(g.s, email) token, expires = await create_magic_link(g.s, user.id) @@ -132,10 +139,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_social_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): @@ -148,17 +155,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_social_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_social_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 diff --git a/federation/bp/identity/routes.py b/federation/bp/identity/routes.py index db8d551..d101c5c 100644 --- a/federation/bp/identity/routes.py +++ b/federation/bp/identity/routes.py @@ -26,6 +26,33 @@ RESERVED = frozenset({ }) +async def _render_choose_username(*, actor=None, error="", username=""): + """Render choose-username page — replaces sx_components helper.""" + from shared.browser.app.csrf import generate_csrf_token + from shared.config import config + from shared.sx.helpers import render_to_sx + from shared.sx.parser import SxExpr + from shared.sx.page import get_template_context + from sxc.pages import _social_page + from markupsafe import escape + + ctx = await get_template_context() + csrf = generate_csrf_token() + ap_domain = config().get("ap_domain", "rose-ash.com") + check_url = url_for("identity.check_username") + + error_sx = await render_to_sx("auth-error-banner", error=error) if error else "" + content = await render_to_sx( + "federation-choose-username", + domain=str(escape(ap_domain)), + error=SxExpr(error_sx) if error_sx else None, + csrf=csrf, username=str(escape(username)), + check_url=check_url, + ) + return await _social_page(ctx, actor, content=content, + title="Choose Username \u2014 Rose Ash") + + def register(url_prefix="/identity"): bp = Blueprint("identity", __name__, url_prefix=url_prefix) @@ -39,11 +66,7 @@ def register(url_prefix="/identity"): if actor: return redirect(url_for("activitypub.actor_profile", username=actor.preferred_username)) - from shared.sx.page import get_template_context - from sx.sx_components import render_choose_username_page - ctx = await get_template_context() - ctx["actor"] = actor - return await render_choose_username_page(ctx) + return await _render_choose_username(actor=actor) @bp.post("/choose-username") async def choose_username(): @@ -71,11 +94,7 @@ def register(url_prefix="/identity"): error = "This username is already taken." if error: - from shared.sx.page import get_template_context - from sx.sx_components import render_choose_username_page - ctx = await get_template_context(error=error, username=username) - ctx["actor"] = None - return await render_choose_username_page(ctx), 400 + return await _render_choose_username(error=error, username=username), 400 # Create ActorProfile with RSA keys display_name = g.user.name or username diff --git a/federation/bp/social/routes.py b/federation/bp/social/routes.py index f3651df..1967955 100644 --- a/federation/bp/social/routes.py +++ b/federation/bp/social/routes.py @@ -7,7 +7,7 @@ from datetime import datetime from quart import Blueprint, request, g, redirect, url_for, abort, Response from shared.services.registry import services -from shared.sx.helpers import sx_response +from shared.sx.helpers import sx_response, render_to_sx log = logging.getLogger(__name__) @@ -47,8 +47,7 @@ def register(url_prefix="/social"): items = await services.federation.get_home_timeline( g.s, actor.id, before=before, ) - from sx.sx_components import render_timeline_items - sx_src = await render_timeline_items(items, "home", actor) + sx_src = await _render_timeline_items(items, "home", actor) return sx_response(sx_src) @bp.get("/public/timeline") @@ -62,8 +61,7 @@ def register(url_prefix="/social"): pass items = await services.federation.get_public_timeline(g.s, before=before) actor = getattr(g, "_social_actor", None) - from sx.sx_components import render_timeline_items - sx_src = await render_timeline_items(items, "public", actor) + sx_src = await _render_timeline_items(items, "public", actor) return sx_response(sx_src) # -- Compose --------------------------------------------------------------- @@ -97,6 +95,8 @@ def register(url_prefix="/social"): @bp.get("/search/page") async def search_page(): + from sxc.pages import _serialize_remote_actor, _serialize_actor + actor = getattr(g, "_social_actor", None) query = request.args.get("q", "").strip() page = request.args.get("page", 1, type=int) @@ -112,8 +112,18 @@ def register(url_prefix="/social"): g.s, actor.preferred_username, page=1, per_page=1000, ) followed_urls = {a.actor_url for a in following} - from sx.sx_components import render_search_results - sx_src = await render_search_results(actors_list, query, page, followed_urls, actor) + + actor_dicts = [_serialize_remote_actor(a) for a in actors_list] + actor_data = _serialize_actor(actor) + parts = [] + for ad in actor_dicts: + parts.append(await render_to_sx("federation-actor-card-from-data", + a=ad, actor=actor_data, + followed_urls=list(followed_urls), list_type="search")) + if len(actors_list) >= 20: + next_url = url_for("social.search_page", q=query, page=page + 1) + parts.append(await render_to_sx("federation-scroll-sentinel", url=next_url)) + sx_src = "(<> " + " ".join(parts) + ")" if parts else "" return sx_response(sx_src) @bp.post("/follow") @@ -144,6 +154,8 @@ def register(url_prefix="/social"): async def _actor_card_response(actor, remote_actor_url, is_followed): """Re-render a single actor card after follow/unfollow via HTMX.""" + from sxc.pages import _serialize_remote_actor, _serialize_actor + remote_dto = await services.federation.get_or_fetch_remote_actor( g.s, remote_actor_url, ) @@ -151,12 +163,12 @@ def register(url_prefix="/social"): return Response("", status=200) followed_urls = {remote_actor_url} if is_followed else set() referer = request.referrer or "" - if "/followers" in referer: - list_type = "followers" - else: - list_type = "following" - from sx.sx_components import render_actor_card - return sx_response(await render_actor_card(remote_dto, actor, followed_urls, list_type=list_type)) + list_type = "followers" if "/followers" in referer else "following" + actor_data = _serialize_actor(actor) + ad = _serialize_remote_actor(remote_dto) + return sx_response(await render_to_sx("federation-actor-card-from-data", + a=ad, actor=actor_data, + followed_urls=list(followed_urls), list_type=list_type)) # -- Interactions ---------------------------------------------------------- @@ -198,7 +210,9 @@ def register(url_prefix="/social"): async def _interaction_buttons_response(actor, object_id, author_inbox): """Re-render interaction buttons after a like/boost action.""" - from shared.models.federation import APInteraction, APRemotePost, APActivity + from shared.models.federation import APInteraction + from shared.browser.app.csrf import generate_csrf_token + from shared.sx.parser import SxExpr from sqlalchemy import select svc = services.federation @@ -242,32 +256,72 @@ def register(url_prefix="/social"): ).limit(1) )).scalar()) - from sx.sx_components import render_interaction_buttons - return sx_response(await render_interaction_buttons( - object_id=object_id, - author_inbox=author_inbox, - like_count=like_count, - boost_count=boost_count, - liked_by_me=liked_by_me, - boosted_by_me=boosted_by_me, - actor=actor, - )) + csrf = generate_csrf_token() + safe_id = object_id.replace("/", "_").replace(":", "_") + target = f"#interactions-{safe_id}" + + if liked_by_me: + like_action = url_for("social.unlike") + like_cls = "text-red-500 hover:text-red-600" + like_icon = "\u2665" + else: + like_action = url_for("social.like") + like_cls = "hover:text-red-500" + like_icon = "\u2661" + + if boosted_by_me: + boost_action = url_for("social.unboost") + boost_cls = "text-green-600 hover:text-green-700" + else: + boost_action = url_for("social.boost") + boost_cls = "hover:text-green-600" + + reply_url = url_for("social.defpage_compose_form", reply_to=object_id) if object_id else "" + reply_sx = await render_to_sx("federation-reply-link", url=reply_url) if reply_url else "" + + like_form = await render_to_sx("federation-like-form", + action=like_action, target=target, oid=object_id, ainbox=author_inbox, + csrf=csrf, cls=f"flex items-center gap-1 {like_cls}", + icon=like_icon, count=str(like_count)) + + boost_form = await render_to_sx("federation-boost-form", + action=boost_action, target=target, oid=object_id, ainbox=author_inbox, + csrf=csrf, cls=f"flex items-center gap-1 {boost_cls}", + count=str(boost_count)) + + return sx_response(await render_to_sx("federation-interaction-buttons", + like=SxExpr(like_form), + boost=SxExpr(boost_form), + reply=SxExpr(reply_sx) if reply_sx else None)) # -- Following / Followers pagination -------------------------------------- @bp.get("/following/page") async def following_list_page(): + from sxc.pages import _serialize_remote_actor, _serialize_actor + actor = _require_actor() page = request.args.get("page", 1, type=int) actors_list, total = await services.federation.get_following( g.s, actor.preferred_username, page=page, ) - from sx.sx_components import render_following_items - sx_src = await render_following_items(actors_list, page, actor) + actor_dicts = [_serialize_remote_actor(a) for a in actors_list] + actor_data = _serialize_actor(actor) + parts = [] + for ad in actor_dicts: + parts.append(await render_to_sx("federation-actor-card-from-data", + a=ad, actor=actor_data, + followed_urls=[], list_type="following")) + if len(actors_list) >= 20: + next_url = url_for("social.following_list_page", page=page + 1) + parts.append(await render_to_sx("federation-scroll-sentinel", url=next_url)) + sx_src = "(<> " + " ".join(parts) + ")" if parts else "" return sx_response(sx_src) @bp.get("/followers/page") async def followers_list_page(): + from sxc.pages import _serialize_remote_actor, _serialize_actor + actor = _require_actor() page = request.args.get("page", 1, type=int) actors_list, total = await services.federation.get_followers_paginated( @@ -277,8 +331,17 @@ def register(url_prefix="/social"): g.s, actor.preferred_username, page=1, per_page=1000, ) followed_urls = {a.actor_url for a in following} - from sx.sx_components import render_followers_items - sx_src = await render_followers_items(actors_list, page, followed_urls, actor) + actor_dicts = [_serialize_remote_actor(a) for a in actors_list] + actor_data = _serialize_actor(actor) + parts = [] + for ad in actor_dicts: + parts.append(await render_to_sx("federation-actor-card-from-data", + a=ad, actor=actor_data, + followed_urls=list(followed_urls), list_type="followers")) + if len(actors_list) >= 20: + next_url = url_for("social.followers_list_page", page=page + 1) + parts.append(await render_to_sx("federation-scroll-sentinel", url=next_url)) + sx_src = "(<> " + " ".join(parts) + ")" if parts else "" return sx_response(sx_src) @bp.get("/actor//timeline") @@ -294,8 +357,7 @@ def register(url_prefix="/social"): items = await services.federation.get_actor_timeline( g.s, id, before=before, ) - from sx.sx_components import render_actor_timeline_items - sx_src = await render_actor_timeline_items(items, id, actor) + sx_src = await _render_timeline_items(items, "actor", actor, id) return sx_response(sx_src) # -- Notifications --------------------------------------------------------- @@ -321,3 +383,26 @@ def register(url_prefix="/social"): return redirect(url_for("defpage_notifications")) return bp + + +async def _render_timeline_items(items, timeline_type, actor, actor_id=None): + """Render timeline pagination items as SX fragment.""" + from sxc.pages import _serialize_timeline_item, _serialize_actor + + item_dicts = [_serialize_timeline_item(i) for i in items] + actor_data = _serialize_actor(actor) + + next_url = None + if items: + last = items[-1] + before = last.published.isoformat() if last.published else "" + if timeline_type == "actor" and actor_id is not None: + next_url = url_for("social.actor_timeline_page", id=actor_id, before=before) + else: + next_url = url_for(f"social.{timeline_type}_timeline_page", before=before) + + return await render_to_sx("federation-timeline-items", + items=item_dicts, + timeline_type=timeline_type, + actor=actor_data, + next_url=next_url) diff --git a/federation/sx/sx_components.py b/federation/sx/sx_components.py deleted file mode 100644 index 46d93f3..0000000 --- a/federation/sx/sx_components.py +++ /dev/null @@ -1,293 +0,0 @@ -""" -Federation service s-expression page components. - -Page helpers now call assembled defcomps in .sx files. This file contains -only functions still called directly from route handlers: full-page renders -(login, choose-username, profile) and POST fragment renderers (interaction -buttons, actor cards, pagination items). -""" -from __future__ import annotations - -import os -from typing import Any -from markupsafe import escape - -from shared.sx.jinja_bridge import load_service_components -from shared.sx.helpers import ( - render_to_sx, - root_header_sx, full_page_sx, header_child_sx, -) - -# Load federation-specific .sx components + handlers at import time -load_service_components(os.path.dirname(os.path.dirname(__file__)), - service_name="federation") - - -# --------------------------------------------------------------------------- -# Serialization helpers (shared with pages/__init__.py) -# --------------------------------------------------------------------------- - -def _serialize_actor(actor) -> dict | 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), - "summary": getattr(actor, "summary", None), - "actor_url": getattr(actor, "actor_url", ""), - "domain": getattr(actor, "domain", ""), - } - - -def _serialize_timeline_item(item) -> dict: - published = getattr(item, "published", None) - return { - "object_id": getattr(item, "object_id", "") or "", - "author_inbox": getattr(item, "author_inbox", "") or "", - "actor_icon": getattr(item, "actor_icon", None), - "actor_name": getattr(item, "actor_name", "?"), - "actor_username": getattr(item, "actor_username", ""), - "actor_domain": getattr(item, "actor_domain", ""), - "content": getattr(item, "content", ""), - "summary": getattr(item, "summary", None), - "published": published.strftime("%b %d, %H:%M") if published else "", - "before_cursor": published.isoformat() if published else "", - "url": getattr(item, "url", None), - "post_type": getattr(item, "post_type", ""), - "boosted_by": getattr(item, "boosted_by", None), - "like_count": getattr(item, "like_count", 0) or 0, - "boost_count": getattr(item, "boost_count", 0) or 0, - "liked_by_me": getattr(item, "liked_by_me", False), - "boosted_by_me": getattr(item, "boosted_by_me", False), - } - - -def _serialize_remote_actor(a) -> dict: - return { - "id": getattr(a, "id", None), - "display_name": getattr(a, "display_name", None) or getattr(a, "preferred_username", ""), - "preferred_username": getattr(a, "preferred_username", ""), - "domain": getattr(a, "domain", ""), - "icon_url": getattr(a, "icon_url", None), - "actor_url": getattr(a, "actor_url", ""), - "summary": getattr(a, "summary", None), - } - - -# --------------------------------------------------------------------------- -# Social page shell -# --------------------------------------------------------------------------- - -async def _social_page(ctx: dict, actor: Any, *, content: str, - title: str = "Rose Ash", meta_html: str = "") -> str: - from shared.sx.parser import SxExpr - actor_data = _serialize_actor(actor) - nav = await render_to_sx("federation-social-nav", actor=actor_data) - social_hdr = await render_to_sx("federation-social-header", nav=SxExpr(nav)) - hdr = await root_header_sx(ctx) - child = await header_child_sx(social_hdr) - header_rows = "(<> " + hdr + " " + child + ")" - return await full_page_sx(ctx, header_rows=header_rows, content=content, - meta_html=meta_html or f'{escape(title)}') - - -# --------------------------------------------------------------------------- -# Public API: Full page renders -# --------------------------------------------------------------------------- - -async def render_federation_home(ctx: dict) -> str: - hdr = await root_header_sx(ctx) - return await full_page_sx(ctx, header_rows=hdr) - - -async def render_login_page(ctx: dict) -> str: - error = ctx.get("error", "") - email = ctx.get("email", "") - content = await render_to_sx("account-login-content", - error=error or None, email=str(escape(email))) - return await _social_page(ctx, None, content=content, title="Login \u2014 Rose Ash") - - -async def render_check_email_page(ctx: dict) -> str: - email = ctx.get("email", "") - email_error = ctx.get("email_error") - content = await render_to_sx("account-check-email-content", - email=str(escape(email)), email_error=email_error) - return await _social_page(ctx, None, content=content, - title="Check your email \u2014 Rose Ash") - - -async def render_choose_username_page(ctx: dict) -> str: - from shared.browser.app.csrf import generate_csrf_token - from quart import url_for - from shared.config import config - from shared.sx.parser import SxExpr - - csrf = generate_csrf_token() - error = ctx.get("error", "") - username = ctx.get("username", "") - ap_domain = config().get("ap_domain", "rose-ash.com") - check_url = url_for("identity.check_username") - actor = ctx.get("actor") - - error_sx = await render_to_sx("auth-error-banner", error=error) if error else "" - content = await render_to_sx( - "federation-choose-username", - domain=str(escape(ap_domain)), - error=SxExpr(error_sx) if error_sx else None, - csrf=csrf, username=str(escape(username)), - check_url=check_url, - ) - return await _social_page(ctx, actor, content=content, - title="Choose Username \u2014 Rose Ash") - - -# --------------------------------------------------------------------------- -# Public API: Pagination fragment renderers -# --------------------------------------------------------------------------- - -async def render_timeline_items(items: list, timeline_type: str, - actor: Any, actor_id: int | None = None) -> str: - from quart import url_for - item_dicts = [_serialize_timeline_item(i) for i in items] - actor_data = _serialize_actor(actor) - - # Build next URL - next_url = None - if items: - last = items[-1] - before = last.published.isoformat() if last.published else "" - if timeline_type == "actor" and actor_id is not None: - next_url = url_for("social.actor_timeline_page", id=actor_id, before=before) - else: - next_url = url_for(f"social.{timeline_type}_timeline_page", before=before) - - return await render_to_sx("federation-timeline-items", - items=item_dicts, - timeline_type=timeline_type, - actor=actor_data, - next_url=next_url) - - -async def render_search_results(actors: list, query: str, page: int, - followed_urls: set, actor: Any) -> str: - from quart import url_for - actor_dicts = [_serialize_remote_actor(a) for a in actors] - actor_data = _serialize_actor(actor) - parts = [] - for ad in actor_dicts: - parts.append(await render_to_sx("federation-actor-card-from-data", - a=ad, - actor=actor_data, - followed_urls=list(followed_urls), - list_type="search")) - if len(actors) >= 20: - next_url = url_for("social.search_page", q=query, page=page + 1) - parts.append(await render_to_sx("federation-scroll-sentinel", url=next_url)) - return "(<> " + " ".join(parts) + ")" if parts else "" - - -async def render_following_items(actors: list, page: int, actor: Any) -> str: - from quart import url_for - actor_dicts = [_serialize_remote_actor(a) for a in actors] - actor_data = _serialize_actor(actor) - parts = [] - for ad in actor_dicts: - parts.append(await render_to_sx("federation-actor-card-from-data", - a=ad, - actor=actor_data, - followed_urls=[], - list_type="following")) - if len(actors) >= 20: - next_url = url_for("social.following_list_page", page=page + 1) - parts.append(await render_to_sx("federation-scroll-sentinel", url=next_url)) - return "(<> " + " ".join(parts) + ")" if parts else "" - - -async def render_followers_items(actors: list, page: int, - followed_urls: set, actor: Any) -> str: - from quart import url_for - actor_dicts = [_serialize_remote_actor(a) for a in actors] - actor_data = _serialize_actor(actor) - parts = [] - for ad in actor_dicts: - parts.append(await render_to_sx("federation-actor-card-from-data", - a=ad, - actor=actor_data, - followed_urls=list(followed_urls), - list_type="followers")) - if len(actors) >= 20: - next_url = url_for("social.followers_list_page", page=page + 1) - parts.append(await render_to_sx("federation-scroll-sentinel", url=next_url)) - return "(<> " + " ".join(parts) + ")" if parts else "" - - -async def render_actor_timeline_items(items: list, actor_id: int, - actor: Any) -> str: - return await render_timeline_items(items, "actor", actor, actor_id) - - -# --------------------------------------------------------------------------- -# Public API: POST handler fragment renderers -# --------------------------------------------------------------------------- - -async def render_interaction_buttons(object_id: str, author_inbox: str, - like_count: int, boost_count: int, - liked_by_me: bool, boosted_by_me: bool, - actor: Any) -> str: - """Render interaction buttons fragment for POST response.""" - from shared.browser.app.csrf import generate_csrf_token - from quart import url_for - from shared.sx.parser import SxExpr - - csrf = generate_csrf_token() - safe_id = object_id.replace("/", "_").replace(":", "_") - target = f"#interactions-{safe_id}" - - if liked_by_me: - like_action = url_for("social.unlike") - like_cls = "text-red-500 hover:text-red-600" - like_icon = "\u2665" - else: - like_action = url_for("social.like") - like_cls = "hover:text-red-500" - like_icon = "\u2661" - - if boosted_by_me: - boost_action = url_for("social.unboost") - boost_cls = "text-green-600 hover:text-green-700" - else: - boost_action = url_for("social.boost") - boost_cls = "hover:text-green-600" - - reply_url = url_for("social.defpage_compose_form", reply_to=object_id) if object_id else "" - reply_sx = await render_to_sx("federation-reply-link", url=reply_url) if reply_url else "" - - like_form = await render_to_sx("federation-like-form", - action=like_action, target=target, oid=object_id, ainbox=author_inbox, - csrf=csrf, cls=f"flex items-center gap-1 {like_cls}", - icon=like_icon, count=str(like_count)) - - boost_form = await render_to_sx("federation-boost-form", - action=boost_action, target=target, oid=object_id, ainbox=author_inbox, - csrf=csrf, cls=f"flex items-center gap-1 {boost_cls}", - count=str(boost_count)) - - return await render_to_sx("federation-interaction-buttons", - like=SxExpr(like_form), - boost=SxExpr(boost_form), - reply=SxExpr(reply_sx) if reply_sx else None) - - -async def render_actor_card(actor_dto: Any, actor: Any, followed_urls: set, - *, list_type: str = "following") -> str: - """Render a single actor card fragment for POST response.""" - actor_data = _serialize_actor(actor) - ad = _serialize_remote_actor(actor_dto) - return await render_to_sx("federation-actor-card-from-data", - a=ad, - actor=actor_data, - followed_urls=list(followed_urls), - list_type=list_type) diff --git a/federation/sxc/pages/__init__.py b/federation/sxc/pages/__init__.py index 1c8ccdf..9442c36 100644 --- a/federation/sxc/pages/__init__.py +++ b/federation/sxc/pages/__init__.py @@ -125,6 +125,23 @@ def _serialize_remote_actor(a) -> dict: } +async def _social_page(ctx: dict, actor, *, content: str, + title: str = "Rose Ash", meta_html: str = "") -> str: + """Build a full social page with social header.""" + from shared.sx.helpers import render_to_sx, root_header_sx, header_child_sx, full_page_sx + from shared.sx.parser import SxExpr + from markupsafe import escape + + actor_data = _serialize_actor(actor) + nav = await render_to_sx("federation-social-nav", actor=actor_data) + social_hdr = await render_to_sx("federation-social-header", nav=SxExpr(nav)) + hdr = await root_header_sx(ctx) + child = await header_child_sx(social_hdr) + header_rows = "(<> " + hdr + " " + child + ")" + return await full_page_sx(ctx, header_rows=header_rows, content=content, + meta_html=meta_html or f'{escape(title)}') + + def _get_actor(): """Return current user's actor or None.""" from quart import g diff --git a/orders/app.py b/orders/app.py index 762b2e1..5f81b9d 100644 --- a/orders/app.py +++ b/orders/app.py @@ -1,7 +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 types import SimpleNamespace @@ -69,10 +67,9 @@ def create_app() -> "Quart": app.jinja_loader, ]) - # Load orders-specific s-expression components (loaded at import time) - import sx.sx_components # noqa: F811 - - # Setup defpage routes + # 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="orders") from sxc.pages import setup_orders_pages setup_orders_pages() diff --git a/orders/bp/checkout/routes.py b/orders/bp/checkout/routes.py index d392df6..8a0929c 100644 --- a/orders/bp/checkout/routes.py +++ b/orders/bp/checkout/routes.py @@ -11,6 +11,118 @@ from services.checkout import validate_webhook_secret, get_order_with_details from services.check_sumup_status import check_sumup_status +async def _render_checkout_return(ctx: dict, order=None, status: str = "", + calendar_entries=None, order_tickets=None) -> str: + """Render checkout return page — replaces sx_components helper.""" + from shared.sx.helpers import ( + render_to_sx, root_header_sx, header_child_sx, full_page_sx, call_url, + ) + from shared.sx.parser import SxExpr + from shared.infrastructure.urls import market_product_url + + filt = await render_to_sx("checkout-return-header", status=status) + + if not order: + content = await render_to_sx("checkout-return-missing") + else: + summary = await render_to_sx("order-summary-card", + order_id=order.id, + created_at=order.created_at.strftime("%-d %b %Y, %H:%M") if order.created_at else None, + description=order.description, status=order.status, + currency=order.currency, + total_amount=f"{order.total_amount:.2f}" if order.total_amount else None, + ) + + items = "" + if order.items: + item_parts = [] + for item in order.items: + product_url = market_product_url(item.product_slug) + if item.product_image: + img = await render_to_sx("order-item-image", + src=item.product_image, + alt=item.product_title or "Product image") + else: + img = await render_to_sx("order-item-no-image") + item_parts.append(await render_to_sx("order-item-row", + href=product_url, img=SxExpr(img), + title=item.product_title or "Unknown product", + pid=f"Product ID: {item.product_id}", + qty=f"Qty: {item.quantity}", + price=f"{item.currency or order.currency or 'GBP'} {item.unit_price or 0:.2f}", + )) + items = await render_to_sx("order-items-panel", + items=SxExpr("(<> " + " ".join(item_parts) + ")")) + + calendar = "" + if calendar_entries: + cal_parts = [] + for e in calendar_entries: + st = e.state or "" + pill = ( + "bg-emerald-100 text-emerald-800" if st == "confirmed" + else "bg-amber-100 text-amber-800" if st == "provisional" + else "bg-blue-100 text-blue-800" if st == "ordered" + else "bg-stone-100 text-stone-700" + ) + ds = e.start_at.strftime("%-d %b %Y, %H:%M") if e.start_at else "" + if e.end_at: + ds += f" \u2013 {e.end_at.strftime('%-d %b %Y, %H:%M')}" + cal_parts.append(await render_to_sx("order-calendar-entry", + name=e.name, + pill=f"inline-flex items-center rounded-full px-2 py-0.5 text-[11px] font-medium {pill}", + status=st.capitalize(), date_str=ds, + cost=f"\u00a3{e.cost or 0:.2f}", + )) + calendar = await render_to_sx("order-calendar-section", + items=SxExpr("(<> " + " ".join(cal_parts) + ")")) + + tickets = "" + if order_tickets: + tk_parts = [] + for tk in order_tickets: + st = tk.state or "" + pill = ( + "bg-emerald-100 text-emerald-800" if st == "confirmed" + else "bg-amber-100 text-amber-800" if st == "reserved" + else "bg-blue-100 text-blue-800" if st == "checked_in" + else "bg-stone-100 text-stone-700" + ) + pill_cls = f"inline-flex items-center rounded-full px-2 py-0.5 text-[11px] font-medium {pill}" + ds = tk.entry_start_at.strftime("%-d %b %Y, %H:%M") if tk.entry_start_at else "" + if tk.entry_end_at: + ds += f" \u2013 {tk.entry_end_at.strftime('%-d %b %Y, %H:%M')}" + tk_parts.append(await render_to_sx("checkout-return-ticket", + name=tk.entry_name, pill=pill_cls, + state=st.replace("_", " ").capitalize(), + type_name=tk.ticket_type_name or None, + date_str=ds, code=tk.code, + price=f"\u00a3{tk.price or 0:.2f}", + )) + tickets = await render_to_sx("checkout-return-tickets", + items=SxExpr("(<> " + " ".join(tk_parts) + ")")) + + status_msg = "" + if order.status == "failed": + status_msg = await render_to_sx("checkout-return-failed", order_id=order.id) + elif order.status == "paid": + status_msg = await render_to_sx("checkout-return-paid") + + content = await render_to_sx("checkout-return-content", + summary=SxExpr(summary), + items=SxExpr(items) if items else None, + calendar=SxExpr(calendar) if calendar else None, + tickets=SxExpr(tickets) if tickets else None, + status_message=SxExpr(status_msg) if status_msg else None, + ) + + account_url = call_url(ctx, "account_url", "") + auth_hdr = await render_to_sx("auth-header-row", account_url=account_url) + hdr = "(<> " + await root_header_sx(ctx) + " " + await header_child_sx(auth_hdr) + ")" + + return await full_page_sx(ctx, header_rows=hdr, filter=filt, content=content) + + def register() -> Blueprint: bp = Blueprint("checkout", __name__, url_prefix="/checkout") @@ -47,12 +159,11 @@ def register() -> Blueprint: async def checkout_return(order_id: int): """Handle the browser returning from SumUp after payment.""" from shared.sx.page import get_template_context - from sx.sx_components import render_checkout_return_page order = await get_order_with_details(g.s, order_id) if not order: tctx = await get_template_context() - html = await render_checkout_return_page(tctx, order=None, status="missing") + html = await _render_checkout_return(tctx, order=None, status="missing") return await make_response(html) if order.page_config_id: @@ -90,7 +201,7 @@ def register() -> Blueprint: await g.s.flush() tctx = await get_template_context() - html = await render_checkout_return_page( + html = await _render_checkout_return( tctx, order=order, status=status, calendar_entries=calendar_entries, order_tickets=order_tickets, diff --git a/orders/bp/order/routes.py b/orders/bp/order/routes.py index ecc1b82..155078d 100644 --- a/orders/bp/order/routes.py +++ b/orders/bp/order/routes.py @@ -70,9 +70,22 @@ def register() -> Blueprint: if not hosted_url: from shared.sx.page import get_template_context - from sx.sx_components import render_checkout_error_page + from shared.sx.helpers import render_to_sx, root_header_sx, header_child_sx, full_page_sx, call_url + from shared.sx.parser import SxExpr + from shared.infrastructure.urls import cart_url tctx = await get_template_context() - html = await render_checkout_error_page(tctx, error="No hosted checkout URL returned from SumUp when trying to reopen payment.", order=order) + account_url = call_url(tctx, "account_url", "") + auth_hdr = await render_to_sx("auth-header-row", account_url=account_url) + hdr = "(<> " + await root_header_sx(tctx) + " " + await header_child_sx(auth_hdr) + ")" + filt = await render_to_sx("checkout-error-header") + order_sx = await render_to_sx("checkout-error-order-id", oid=f"#{order.id}") + content = await render_to_sx( + "checkout-error-content", + msg="No hosted checkout URL returned from SumUp when trying to reopen payment.", + order=SxExpr(order_sx), + back_url=cart_url("/"), + ) + html = await full_page_sx(tctx, header_rows=hdr, filter=filt, content=content) return await make_response(html, 500) return redirect(hosted_url) diff --git a/orders/sx/sx_components.py b/orders/sx/sx_components.py deleted file mode 100644 index be7f1cc..0000000 --- a/orders/sx/sx_components.py +++ /dev/null @@ -1,189 +0,0 @@ -""" -Orders service s-expression page components. - -Checkout error/return pages are still rendered from Python because they -use ``full_page_sx()`` with custom layouts. All other order rendering -is now handled by .sx defcomps. -""" -from __future__ import annotations - -import os -from typing import Any - -from shared.sx.jinja_bridge import load_service_components -from shared.sx.helpers import ( - call_url, render_to_sx, - root_header_sx, full_page_sx, header_child_sx, -) -from shared.infrastructure.urls import market_product_url, cart_url - -# Load orders-specific .sx components + handlers at import time -load_service_components(os.path.dirname(os.path.dirname(__file__)), - service_name="orders") - - -# --------------------------------------------------------------------------- -# Public API: Checkout error -# --------------------------------------------------------------------------- - -async def render_checkout_error_page(ctx: dict, error: str | None = None, order: Any | None = None) -> str: - """Full page: checkout error (sx wire format).""" - account_url = call_url(ctx, "account_url", "") - auth_hdr = await render_to_sx("auth-header-row", account_url=account_url) - hdr = await root_header_sx(ctx) - hdr = "(<> " + hdr + " " + await header_child_sx(auth_hdr) + ")" - filt = await render_to_sx("checkout-error-header") - - err_msg = error or "Unexpected error while creating the hosted checkout session." - order_sx = "" - if order: - order_sx = await render_to_sx("checkout-error-order-id", oid=f"#{order.id}") - from shared.sx.parser import SxExpr - content = await render_to_sx( - "checkout-error-content", - msg=err_msg, - order=SxExpr(order_sx) if order_sx else None, - back_url=cart_url("/"), - ) - - return await full_page_sx(ctx, header_rows=hdr, filter=filt, content=content) - - -# --------------------------------------------------------------------------- -# Public API: Checkout return -# --------------------------------------------------------------------------- - -async def render_checkout_return_page(ctx: dict, order: Any | None, - status: str, - calendar_entries: list | None = None, - order_tickets: list | None = None) -> str: - """Full page: checkout return after SumUp payment (sx wire format).""" - from shared.sx.parser import SxExpr - - filt = await render_to_sx("checkout-return-header", status=status) - - if not order: - content = await render_to_sx("checkout-return-missing") - else: - # Serialize order data for defcomp - order_dict = { - "id": order.id, - "status": order.status or "pending", - "created_at_formatted": order.created_at.strftime("%-d %b %Y, %H:%M") if order.created_at else None, - "description": order.description, - "currency": order.currency, - "total_formatted": f"{order.total_amount:.2f}" if order.total_amount else None, - "items": [ - { - "product_url": market_product_url(item.product_slug), - "product_image": item.product_image, - "product_title": item.product_title, - "product_id": item.product_id, - "quantity": item.quantity, - "currency": item.currency, - "unit_price_formatted": f"{item.unit_price or 0:.2f}", - } - for item in (order.items or []) - ], - } - - summary = await render_to_sx("order-summary-card", - order_id=order.id, - created_at=order_dict["created_at_formatted"], - description=order.description, status=order.status, - currency=order.currency, - total_amount=order_dict["total_formatted"], - ) - - # Items - items = "" - if order.items: - item_parts = [] - for item_d in order_dict["items"]: - if item_d["product_image"]: - img = await render_to_sx("order-item-image", - src=item_d["product_image"], - alt=item_d["product_title"] or "Product image") - else: - img = await render_to_sx("order-item-no-image") - item_parts.append(await render_to_sx("order-item-row", - href=item_d["product_url"], img=SxExpr(img), - title=item_d["product_title"] or "Unknown product", - pid=f"Product ID: {item_d['product_id']}", - qty=f"Qty: {item_d['quantity']}", - price=f"{item_d['currency'] or order.currency or 'GBP'} {item_d['unit_price_formatted']}", - )) - items = await render_to_sx("order-items-panel", - items=SxExpr("(<> " + " ".join(item_parts) + ")")) - - # Calendar entries - calendar = "" - if calendar_entries: - cal_parts = [] - for e in calendar_entries: - st = e.state or "" - pill = ( - "bg-emerald-100 text-emerald-800" if st == "confirmed" - else "bg-amber-100 text-amber-800" if st == "provisional" - else "bg-blue-100 text-blue-800" if st == "ordered" - else "bg-stone-100 text-stone-700" - ) - ds = e.start_at.strftime("%-d %b %Y, %H:%M") if e.start_at else "" - if e.end_at: - ds += f" \u2013 {e.end_at.strftime('%-d %b %Y, %H:%M')}" - cal_parts.append(await render_to_sx("order-calendar-entry", - name=e.name, - pill=f"inline-flex items-center rounded-full px-2 py-0.5 text-[11px] font-medium {pill}", - status=st.capitalize(), date_str=ds, - cost=f"\u00a3{e.cost or 0:.2f}", - )) - calendar = await render_to_sx("order-calendar-section", - items=SxExpr("(<> " + " ".join(cal_parts) + ")")) - - # Tickets - tickets = "" - if order_tickets: - tk_parts = [] - for tk in order_tickets: - st = tk.state or "" - pill = ( - "bg-emerald-100 text-emerald-800" if st == "confirmed" - else "bg-amber-100 text-amber-800" if st == "reserved" - else "bg-blue-100 text-blue-800" if st == "checked_in" - else "bg-stone-100 text-stone-700" - ) - pill_cls = f"inline-flex items-center rounded-full px-2 py-0.5 text-[11px] font-medium {pill}" - ds = tk.entry_start_at.strftime("%-d %b %Y, %H:%M") if tk.entry_start_at else "" - if tk.entry_end_at: - ds += f" \u2013 {tk.entry_end_at.strftime('%-d %b %Y, %H:%M')}" - tk_parts.append(await render_to_sx("checkout-return-ticket", - name=tk.entry_name, pill=pill_cls, - state=st.replace("_", " ").capitalize(), - type_name=tk.ticket_type_name or None, - date_str=ds, code=tk.code, - price=f"\u00a3{tk.price or 0:.2f}", - )) - tickets = await render_to_sx("checkout-return-tickets", - items=SxExpr("(<> " + " ".join(tk_parts) + ")")) - - # Status message - status_msg = "" - if order.status == "failed": - status_msg = await render_to_sx("checkout-return-failed", order_id=order.id) - elif order.status == "paid": - status_msg = await render_to_sx("checkout-return-paid") - - content = await render_to_sx("checkout-return-content", - summary=SxExpr(summary), - items=SxExpr(items) if items else None, - calendar=SxExpr(calendar) if calendar else None, - tickets=SxExpr(tickets) if tickets else None, - status_message=SxExpr(status_msg) if status_msg else None, - ) - - account_url = call_url(ctx, "account_url", "") - auth_hdr = await render_to_sx("auth-header-row", account_url=account_url) - hdr = await root_header_sx(ctx) - hdr = "(<> " + hdr + " " + await header_child_sx(auth_hdr) + ")" - - return await full_page_sx(ctx, header_rows=hdr, filter=filt, content=content) diff --git a/orders/tests/test_sx_helpers.py b/orders/tests/test_sx_helpers.py deleted file mode 100644 index d8420d3..0000000 --- a/orders/tests/test_sx_helpers.py +++ /dev/null @@ -1,32 +0,0 @@ -"""Unit tests for orders sx component helpers.""" -from __future__ import annotations - -import pytest - -from orders.sx.sx_components import _status_pill_cls - - -class TestStatusPillCls: - def test_paid(self): - result = _status_pill_cls("paid") - assert "emerald" in result - - def test_Paid_uppercase(self): - result = _status_pill_cls("Paid") - assert "emerald" in result - - def test_failed(self): - result = _status_pill_cls("failed") - assert "rose" in result - - def test_cancelled(self): - result = _status_pill_cls("cancelled") - assert "rose" in result - - def test_pending(self): - result = _status_pill_cls("pending") - assert "stone" in result - - def test_unknown(self): - result = _status_pill_cls("refunded") - assert "stone" in result