Auto-mount fragment handlers: eliminate fragment blueprint boilerplate across all 8 services
Fragment read API is now fully declarative — every handler is a defhandler s-expression dispatched through one shared auto_mount_fragment_handlers() function. Replaces 8 near-identical blueprint files (~35 lines each) with a single function call per service. Events Python handlers (container-cards, account-page) extracted to a standalone module. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -9,7 +9,7 @@ from jinja2 import FileSystemLoader, ChoiceLoader
|
||||
|
||||
from shared.infrastructure.factory import create_base_app
|
||||
|
||||
from bp import register_all_events, register_calendar, register_calendars, register_markets, register_page, register_fragments, register_actions, register_data
|
||||
from bp import register_all_events, register_calendar, register_calendars, register_markets, register_page, register_actions, register_data
|
||||
|
||||
|
||||
async def events_context() -> dict:
|
||||
@@ -112,7 +112,12 @@ def create_app() -> "Quart":
|
||||
url_prefix="/<slug>/markets",
|
||||
)
|
||||
|
||||
app.register_blueprint(register_fragments())
|
||||
from shared.sx.handlers import auto_mount_fragment_handlers
|
||||
from bp.fragments.python_handlers import container_cards_handler, account_page_handler
|
||||
add_fragment_handler = auto_mount_fragment_handlers(app, "events")
|
||||
add_fragment_handler("container-cards", container_cards_handler, content_type="text/html")
|
||||
add_fragment_handler("account-page", account_page_handler)
|
||||
|
||||
app.register_blueprint(register_actions())
|
||||
app.register_blueprint(register_data())
|
||||
|
||||
|
||||
@@ -3,6 +3,5 @@ from .calendar.routes import register as register_calendar
|
||||
from .calendars.routes import register as register_calendars
|
||||
from .markets.routes import register as register_markets
|
||||
from .page.routes import register as register_page
|
||||
from .fragments import register_fragments
|
||||
from .actions import register_actions
|
||||
from .data import register_data
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
from .routes import register as register_fragments
|
||||
|
||||
58
events/bp/fragments/python_handlers.py
Normal file
58
events/bp/fragments/python_handlers.py
Normal file
@@ -0,0 +1,58 @@
|
||||
"""Python fragment handlers for events.
|
||||
|
||||
These handlers call domain services and use sx_call() for rendering,
|
||||
so they can't be expressed as declarative .sx handlers.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from quart import g, request
|
||||
|
||||
from shared.services.registry import services
|
||||
|
||||
|
||||
async def container_cards_handler():
|
||||
"""Container-cards fragment: entries for blog listing cards.
|
||||
|
||||
Returns text/html with <!-- card-widget:POST_ID --> comment markers
|
||||
so the blog consumer can split per-post fragments.
|
||||
"""
|
||||
from sx.sx_components import render_fragment_container_cards
|
||||
|
||||
post_ids_raw = request.args.get("post_ids", "")
|
||||
post_slugs_raw = request.args.get("post_slugs", "")
|
||||
post_ids = [int(x) for x in post_ids_raw.split(",") if x.strip()]
|
||||
post_slugs = [x.strip() for x in post_slugs_raw.split(",") if x.strip()]
|
||||
if not post_ids:
|
||||
return ""
|
||||
|
||||
slug_map = {}
|
||||
for i, pid in enumerate(post_ids):
|
||||
slug_map[pid] = post_slugs[i] if i < len(post_slugs) else ""
|
||||
|
||||
batch = await services.calendar.confirmed_entries_for_posts(g.s, post_ids)
|
||||
return render_fragment_container_cards(batch, post_ids, slug_map)
|
||||
|
||||
|
||||
async def account_page_handler():
|
||||
"""Account-page fragment: tickets or bookings panel.
|
||||
|
||||
Returns text/sx — the account app embeds this as sx source.
|
||||
"""
|
||||
from sx.sx_components import (
|
||||
render_fragment_account_tickets,
|
||||
render_fragment_account_bookings,
|
||||
)
|
||||
|
||||
slug = request.args.get("slug", "")
|
||||
user_id = request.args.get("user_id", type=int)
|
||||
if not user_id:
|
||||
return ""
|
||||
|
||||
if slug == "tickets":
|
||||
tickets = await services.calendar.user_tickets(g.s, user_id=user_id)
|
||||
return render_fragment_account_tickets(tickets)
|
||||
elif slug == "bookings":
|
||||
bookings = await services.calendar.user_bookings(g.s, user_id=user_id)
|
||||
return render_fragment_account_bookings(bookings)
|
||||
return ""
|
||||
@@ -1,104 +0,0 @@
|
||||
"""Events app fragment endpoints.
|
||||
|
||||
Exposes sx fragments at ``/internal/fragments/<type>`` for consumption
|
||||
by other coop apps via the fragment client.
|
||||
|
||||
All handlers are defined declaratively in .sx files under
|
||||
``events/sx/handlers/`` and dispatched via the sx handler registry.
|
||||
|
||||
container-cards and account-page remain as Python handlers because they
|
||||
call domain service methods and return batched/conditional content, but
|
||||
they use sx_call() for rendering (no Jinja templates).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from quart import Blueprint, Response, g, request
|
||||
|
||||
from shared.infrastructure.fragments import FRAGMENT_HEADER
|
||||
from shared.services.registry import services
|
||||
from shared.sx.handlers import get_handler, execute_handler
|
||||
|
||||
|
||||
def register():
|
||||
bp = Blueprint("fragments", __name__, url_prefix="/internal/fragments")
|
||||
|
||||
_handlers: dict[str, object] = {}
|
||||
|
||||
# Fragment types that return HTML (comment-delimited batch)
|
||||
_html_types = {"container-cards"}
|
||||
|
||||
@bp.before_request
|
||||
async def _require_fragment_header():
|
||||
if not request.headers.get(FRAGMENT_HEADER):
|
||||
return Response("", status=403)
|
||||
|
||||
@bp.get("/<fragment_type>")
|
||||
async def get_fragment(fragment_type: str):
|
||||
# 1. Check Python handlers first
|
||||
handler = _handlers.get(fragment_type)
|
||||
if handler is not None:
|
||||
result = await handler()
|
||||
ct = "text/html" if fragment_type in _html_types else "text/sx"
|
||||
return Response(result, status=200, content_type=ct)
|
||||
|
||||
# 2. Check sx handler registry
|
||||
handler_def = get_handler("events", fragment_type)
|
||||
if handler_def is not None:
|
||||
result = await execute_handler(
|
||||
handler_def, "events", args=dict(request.args),
|
||||
)
|
||||
return Response(result, status=200, content_type="text/sx")
|
||||
|
||||
return Response("", status=200, content_type="text/sx")
|
||||
|
||||
# --- container-cards fragment: entries for blog listing cards -----------
|
||||
# Returns text/html with <!-- card-widget:POST_ID --> comment markers
|
||||
# so the blog consumer can split per-post fragments.
|
||||
|
||||
async def _container_cards_handler():
|
||||
from sx.sx_components import render_fragment_container_cards
|
||||
|
||||
post_ids_raw = request.args.get("post_ids", "")
|
||||
post_slugs_raw = request.args.get("post_slugs", "")
|
||||
post_ids = [int(x) for x in post_ids_raw.split(",") if x.strip()]
|
||||
post_slugs = [x.strip() for x in post_slugs_raw.split(",") if x.strip()]
|
||||
if not post_ids:
|
||||
return ""
|
||||
|
||||
slug_map = {}
|
||||
for i, pid in enumerate(post_ids):
|
||||
slug_map[pid] = post_slugs[i] if i < len(post_slugs) else ""
|
||||
|
||||
batch = await services.calendar.confirmed_entries_for_posts(g.s, post_ids)
|
||||
return render_fragment_container_cards(batch, post_ids, slug_map)
|
||||
|
||||
_handlers["container-cards"] = _container_cards_handler
|
||||
|
||||
# --- account-page fragment: tickets or bookings panel ------------------
|
||||
# Returns text/sx — the account app embeds this as sx source.
|
||||
|
||||
async def _account_page_handler():
|
||||
from sx.sx_components import (
|
||||
render_fragment_account_tickets,
|
||||
render_fragment_account_bookings,
|
||||
)
|
||||
|
||||
slug = request.args.get("slug", "")
|
||||
user_id = request.args.get("user_id", type=int)
|
||||
if not user_id:
|
||||
return ""
|
||||
|
||||
if slug == "tickets":
|
||||
tickets = await services.calendar.user_tickets(g.s, user_id=user_id)
|
||||
return render_fragment_account_tickets(tickets)
|
||||
elif slug == "bookings":
|
||||
bookings = await services.calendar.user_bookings(g.s, user_id=user_id)
|
||||
return render_fragment_account_bookings(bookings)
|
||||
return ""
|
||||
|
||||
_handlers["account-page"] = _account_page_handler
|
||||
|
||||
bp._fragment_handlers = _handlers
|
||||
|
||||
return bp
|
||||
Reference in New Issue
Block a user