art-dag/mono
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
57a31a3b83f3cac9ef651feb9b92dc48ab1135bf
mono/events/sxc/pages/helpers.py
giles 57a31a3b83 Convert all 23 register_custom_layout calls to register_sx_layout across 6 services 
Layout defcomps are now fully self-contained via IO-primitive auto-fetch
macros, eliminating Python layout functions that manually threaded context
values through SxExpr wrappers.

Services converted:
- Federation (1 layout): social
- Blog (7 layouts): blog, blog-settings, blog-cache, blog-snippets,
  blog-menu-items, blog-tag-groups, blog-tag-group-edit
- SX docs (2 layouts): sx, sx-section
- Cart (2 layouts): cart-page, cart-admin + orders/order-detail
- Events (9 layouts): calendar-admin, slots, slot, day-admin, entry,
  entry-admin, ticket-types, ticket-type, markets
- Market (2 layouts): market, market-admin

New IO primitives added to shared/sx/primitives_io.py:
- federation-actor-ctx, cart-page-ctx, request-view-args
- events-calendar-ctx, events-day-ctx, events-entry-ctx,
  events-slot-ctx, events-ticket-type-ctx
- market-header-ctx (pre-builds desktop/mobile nav as SxExpr)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-04 22:21:44 +00:00

469 lines
17 KiB
Python
Raw Blame History

"""Layout registrations, page helpers, and shared hydration helpers."""
from __future__ import annotations
from typing import Any
from shared.sx.helpers import sx_call
from .calendar import (
_calendar_admin_main_panel_html,
_day_admin_main_panel_html,
_markets_main_panel_html,
)
from .entries import (
_entry_main_panel_html,
_entry_nav_html,
_entry_admin_main_panel_html,
)
from .tickets import (
_tickets_main_panel_html, _ticket_detail_panel_html,
_ticket_admin_main_panel_html,
render_ticket_type_main_panel, render_ticket_types_table,
)
from .slots import render_slot_main_panel, render_slots_table
# ---------------------------------------------------------------------------
# Shared hydration helpers
# ---------------------------------------------------------------------------
def _add_to_defpage_ctx(**kwargs: Any) -> None:
"""Add data to g._defpage_ctx for the app-level context_processor."""
from quart import g
if not hasattr(g, '_defpage_ctx'):
g._defpage_ctx = {}
g._defpage_ctx.update(kwargs)
def _ensure_post_defpage_ctx() -> None:
"""Copy g.post_data["post"] into g._defpage_ctx for layout IO primitives."""
from quart import g
post_data = getattr(g, "post_data", None)
if post_data and post_data.get("post"):
_add_to_defpage_ctx(post=post_data["post"])
async def _ensure_container_nav_defpage_ctx() -> None:
"""Fetch container_nav and add to g._defpage_ctx for layout IO primitives."""
from quart import g
dctx = getattr(g, "_defpage_ctx", None) or {}
if dctx.get("container_nav"):
return
post = dctx.get("post") or {}
post_id = post.get("id")
slug = post.get("slug", "")
if not post_id:
return
from shared.infrastructure.fragments import fetch_fragments
current_cal = getattr(g, "calendar_slug", "") or ""
nav_params = {
"container_type": "page",
"container_id": str(post_id),
"post_slug": slug,
"current_calendar": current_cal,
}
events_nav, market_nav = await fetch_fragments([
("events", "container-nav", nav_params),
("market", "container-nav", nav_params),
], required=False)
_add_to_defpage_ctx(container_nav=events_nav + market_nav)
async def _ensure_calendar(calendar_slug: str | None) -> None:
"""Load calendar into g.calendar if not already present."""
from quart import g, abort
if hasattr(g, 'calendar'):
_add_to_defpage_ctx(calendar=g.calendar)
return
from bp.calendar.services.calendar_view import (
get_calendar_by_post_and_slug, get_calendar_by_slug,
)
post_data = getattr(g, "post_data", None)
if post_data:
post_id = (post_data.get("post") or {}).get("id")
cal = await get_calendar_by_post_and_slug(g.s, post_id, calendar_slug)
else:
cal = await get_calendar_by_slug(g.s, calendar_slug)
if not cal:
abort(404)
g.calendar = cal
g.calendar_slug = calendar_slug
_add_to_defpage_ctx(calendar=cal)
_ensure_post_defpage_ctx()
async def _ensure_entry(entry_id: int | None) -> None:
"""Load calendar entry into g.entry if not already present."""
from quart import g, abort
if hasattr(g, 'entry'):
_add_to_defpage_ctx(entry=g.entry)
return
from sqlalchemy import select
from models.calendars import CalendarEntry
result = await g.s.execute(
select(CalendarEntry).where(
CalendarEntry.id == entry_id,
CalendarEntry.deleted_at.is_(None),
)
)
entry = result.scalar_one_or_none()
if entry is None:
abort(404)
g.entry = entry
_add_to_defpage_ctx(entry=entry)
async def _ensure_entry_context(entry_id: int | None) -> None:
"""Load full entry context (ticket data, posts) into g.* and _defpage_ctx."""
from quart import g
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from models.calendars import CalendarEntry
from bp.tickets.services.tickets import (
get_available_ticket_count,
get_sold_ticket_count,
get_user_reserved_count,
)
from shared.infrastructure.cart_identity import current_cart_identity
from bp.calendar_entry.services.post_associations import get_entry_posts
await _ensure_entry(entry_id)
# Reload with ticket_types eagerly loaded
stmt = (
select(CalendarEntry)
.where(CalendarEntry.id == entry_id, CalendarEntry.deleted_at.is_(None))
.options(selectinload(CalendarEntry.ticket_types))
)
result = await g.s.execute(stmt)
calendar_entry = result.scalar_one_or_none()
if calendar_entry and getattr(g, "calendar", None):
if calendar_entry.calendar_id != g.calendar.id:
calendar_entry = None
if calendar_entry:
await g.s.refresh(calendar_entry, ['slot'])
g.entry = calendar_entry
entry_posts = await get_entry_posts(g.s, calendar_entry.id)
ticket_remaining = await get_available_ticket_count(g.s, calendar_entry.id)
ticket_sold_count = await get_sold_ticket_count(g.s, calendar_entry.id)
ident = current_cart_identity()
user_ticket_count = await get_user_reserved_count(
g.s, calendar_entry.id,
user_id=ident["user_id"],
session_id=ident["session_id"],
)
user_ticket_counts_by_type = {}
if calendar_entry.ticket_types:
for tt in calendar_entry.ticket_types:
if tt.deleted_at is None:
user_ticket_counts_by_type[tt.id] = await get_user_reserved_count(
g.s, calendar_entry.id,
user_id=ident["user_id"],
session_id=ident["session_id"],
ticket_type_id=tt.id,
)
_add_to_defpage_ctx(
entry=calendar_entry,
entry_posts=entry_posts,
ticket_remaining=ticket_remaining,
ticket_sold_count=ticket_sold_count,
user_ticket_count=user_ticket_count,
user_ticket_counts_by_type=user_ticket_counts_by_type,
)
async def _ensure_day_data(year: int, month: int, day: int) -> None:
"""Load day-specific data for layout header functions."""
from quart import g, session as qsession
if hasattr(g, 'day_date'):
return
from datetime import date as date_cls, datetime, timezone, timedelta
from sqlalchemy import select
from bp.calendar.services import get_visible_entries_for_period
from models.calendars import CalendarSlot
calendar = getattr(g, "calendar", None)
if not calendar:
return
try:
day_date = date_cls(year, month, day)
except (ValueError, TypeError):
return
period_start = datetime(year, month, day, tzinfo=timezone.utc)
period_end = period_start + timedelta(days=1)
user = getattr(g, "user", None)
session_id = qsession.get("calendar_sid")
visible = await get_visible_entries_for_period(
sess=g.s,
calendar_id=calendar.id,
period_start=period_start,
period_end=period_end,
user=user,
session_id=session_id,
)
weekday_attr = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"][day_date.weekday()]
stmt = (
select(CalendarSlot)
.where(
CalendarSlot.calendar_id == calendar.id,
getattr(CalendarSlot, weekday_attr) == True, # noqa: E712
CalendarSlot.deleted_at.is_(None),
)
.order_by(CalendarSlot.time_start.asc(), CalendarSlot.id.asc())
)
result = await g.s.execute(stmt)
day_slots = list(result.scalars())
g.day_date = day_date
_add_to_defpage_ctx(
qsession=qsession,
day_date=day_date,
day=day,
year=year,
month=month,
day_entries=visible.merged_entries,
user_entries=visible.user_entries,
confirmed_entries=visible.confirmed_entries,
day_slots=day_slots,
)
# ---------------------------------------------------------------------------
# Layouts — all layouts delegate to .sx defcomps via register_sx_layout
# ---------------------------------------------------------------------------
def _register_events_layouts() -> None:
from shared.sx.layouts import register_sx_layout
register_sx_layout("events-calendar-admin",
"events-cal-admin-layout-full", "events-cal-admin-layout-oob")
register_sx_layout("events-slots",
"events-slots-layout-full", "events-slots-layout-oob")
register_sx_layout("events-slot",
"events-slot-layout-full", "events-slot-layout-oob")
register_sx_layout("events-day-admin",
"events-day-admin-layout-full", "events-day-admin-layout-oob")
register_sx_layout("events-entry",
"events-entry-layout-full", "events-entry-layout-oob")
register_sx_layout("events-entry-admin",
"events-entry-admin-layout-full", "events-entry-admin-layout-oob")
register_sx_layout("events-ticket-types",
"events-ticket-types-layout-full", "events-ticket-types-layout-oob")
register_sx_layout("events-ticket-type",
"events-ticket-type-layout-full", "events-ticket-type-layout-oob")
register_sx_layout("events-markets",
"events-markets-layout-full", "events-markets-layout-oob")
# ---------------------------------------------------------------------------
# Page helpers
# ---------------------------------------------------------------------------
def _register_events_helpers() -> None:
from shared.sx.pages import register_page_helpers
register_page_helpers("events", {
"calendar-admin-content": _h_calendar_admin_content,
"day-admin-content": _h_day_admin_content,
"slots-content": _h_slots_content,
"slot-content": _h_slot_content,
"entry-content": _h_entry_content,
"entry-menu": _h_entry_menu,
"entry-admin-content": _h_entry_admin_content,
"admin-menu": _h_admin_menu,
"ticket-types-content": _h_ticket_types_content,
"ticket-type-content": _h_ticket_type_content,
"tickets-content": _h_tickets_content,
"ticket-detail-content": _h_ticket_detail_content,
"ticket-admin-content": _h_ticket_admin_content,
"markets-content": _h_markets_content,
})
async def _h_calendar_admin_content(calendar_slug=None, **kw):
await _ensure_calendar(calendar_slug)
await _ensure_container_nav_defpage_ctx()
from shared.sx.page import get_template_context
ctx = await get_template_context()
return _calendar_admin_main_panel_html(ctx)
async def _h_day_admin_content(calendar_slug=None, year=None, month=None, day=None, **kw):
await _ensure_calendar(calendar_slug)
await _ensure_container_nav_defpage_ctx()
if year is not None:
await _ensure_day_data(int(year), int(month), int(day))
return _day_admin_main_panel_html({})
async def _h_slots_content(calendar_slug=None, **kw):
from quart import g
await _ensure_calendar(calendar_slug)
await _ensure_container_nav_defpage_ctx()
calendar = getattr(g, "calendar", None)
from bp.slots.services.slots import list_slots as svc_list_slots
slots = await svc_list_slots(g.s, calendar.id) if calendar else []
_add_to_defpage_ctx(slots=slots)
return render_slots_table(slots, calendar)
async def _h_slot_content(calendar_slug=None, slot_id=None, **kw):
from quart import g, abort
await _ensure_calendar(calendar_slug)
await _ensure_container_nav_defpage_ctx()
from bp.slot.services.slot import get_slot as svc_get_slot
slot = await svc_get_slot(g.s, slot_id) if slot_id else None
if not slot:
abort(404)
g.slot = slot
_add_to_defpage_ctx(slot=slot)
calendar = getattr(g, "calendar", None)
return render_slot_main_panel(slot, calendar)
async def _h_entry_content(calendar_slug=None, entry_id=None, **kw):
await _ensure_calendar(calendar_slug)
await _ensure_entry_context(entry_id)
from shared.sx.page import get_template_context
ctx = await get_template_context()
return _entry_main_panel_html(ctx)
async def _h_entry_menu(calendar_slug=None, entry_id=None, **kw):
await _ensure_calendar(calendar_slug)
await _ensure_entry_context(entry_id)
from shared.sx.page import get_template_context
ctx = await get_template_context()
return _entry_nav_html(ctx)
async def _h_entry_admin_content(calendar_slug=None, entry_id=None, **kw):
await _ensure_calendar(calendar_slug)
await _ensure_container_nav_defpage_ctx()
await _ensure_entry_context(entry_id)
from shared.sx.page import get_template_context
ctx = await get_template_context()
return _entry_admin_main_panel_html(ctx)
def _h_admin_menu():
return sx_call("events-admin-placeholder-nav")
async def _h_ticket_types_content(calendar_slug=None, entry_id=None,
year=None, month=None, day=None, **kw):
from quart import g
await _ensure_calendar(calendar_slug)
await _ensure_entry(entry_id)
entry = getattr(g, "entry", None)
calendar = getattr(g, "calendar", None)
from bp.ticket_types.services.tickets import list_ticket_types as svc_list_ticket_types
ticket_types = await svc_list_ticket_types(g.s, entry.id) if entry else []
_add_to_defpage_ctx(ticket_types=ticket_types)
return render_ticket_types_table(ticket_types, entry, calendar, day, month, year)
async def _h_ticket_type_content(calendar_slug=None, entry_id=None,
ticket_type_id=None, year=None, month=None, day=None, **kw):
from quart import g, abort
await _ensure_calendar(calendar_slug)
await _ensure_entry(entry_id)
from bp.ticket_type.services.ticket import get_ticket_type as svc_get_ticket_type
ticket_type = await svc_get_ticket_type(g.s, ticket_type_id) if ticket_type_id else None
if not ticket_type:
abort(404)
g.ticket_type = ticket_type
_add_to_defpage_ctx(ticket_type=ticket_type)
entry = getattr(g, "entry", None)
calendar = getattr(g, "calendar", None)
return render_ticket_type_main_panel(ticket_type, entry, calendar, day, month, year)
async def _h_tickets_content(**kw):
from quart import g
from shared.infrastructure.cart_identity import current_cart_identity
from bp.tickets.services.tickets import get_user_tickets
ident = current_cart_identity()
tickets = await get_user_tickets(
g.s,
user_id=ident["user_id"],
session_id=ident["session_id"],
)
from shared.sx.page import get_template_context
ctx = await get_template_context()
return _tickets_main_panel_html(ctx, tickets)
async def _h_ticket_detail_content(code=None, **kw):
from quart import g, abort
from shared.infrastructure.cart_identity import current_cart_identity
from bp.tickets.services.tickets import get_ticket_by_code
ticket = await get_ticket_by_code(g.s, code) if code else None
if not ticket:
abort(404)
# Verify ownership
ident = current_cart_identity()
if ident["user_id"] is not None:
if ticket.user_id != ident["user_id"]:
abort(404)
elif ident["session_id"] is not None:
if ticket.session_id != ident["session_id"]:
abort(404)
else:
abort(404)
from shared.sx.page import get_template_context
ctx = await get_template_context()
return _ticket_detail_panel_html(ctx, ticket)
async def _h_ticket_admin_content(**kw):
from quart import g
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
from models.calendars import CalendarEntry, Ticket
result = await g.s.execute(
select(Ticket)
.options(
selectinload(Ticket.entry).selectinload(CalendarEntry.calendar),
selectinload(Ticket.ticket_type),
)
.order_by(Ticket.created_at.desc())
.limit(50)
)
tickets = result.scalars().all()
total = await g.s.scalar(select(func.count(Ticket.id)))
confirmed = await g.s.scalar(
select(func.count(Ticket.id)).where(Ticket.state == "confirmed")
)
checked_in = await g.s.scalar(
select(func.count(Ticket.id)).where(Ticket.state == "checked_in")
)
reserved = await g.s.scalar(
select(func.count(Ticket.id)).where(Ticket.state == "reserved")
)
stats = {
"total": total or 0,
"confirmed": confirmed or 0,
"checked_in": checked_in or 0,
"reserved": reserved or 0,
}
from shared.sx.page import get_template_context
ctx = await get_template_context()
return _ticket_admin_main_panel_html(ctx, tickets, stats)
async def _h_markets_content(**kw):
_ensure_post_defpage_ctx()
from shared.sx.page import get_template_context
ctx = await get_template_context()
return _markets_main_panel_html(ctx)
Reference in New Issue View Git Blame Copy Permalink