Auto-mount defpages: eliminate Python route stubs across all 9 services
Some checks failed
Build and Deploy / build-and-deploy (push) Failing after 16s

Defpages are now declared with absolute paths in .sx files and auto-mounted
directly on the Quart app, removing ~850 lines of blueprint mount_pages calls,
before_request hooks, and g.* wrapper boilerplate. A new page = one defpage
declaration, nothing else.

Infrastructure:
- async_eval awaits coroutine results from callable dispatch
- auto_mount_pages() mounts all registered defpages on the app
- g._defpage_ctx pattern passes helper data to layout context

Migrated: sx, account, orders, federation, cart, market, events, blog

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-03 19:03:15 +00:00
parent 4ba63bda17
commit 293f7713d6
63 changed files with 1340 additions and 1216 deletions

View File

@@ -17,6 +17,96 @@ def _load_blog_page_files() -> None:
load_page_dir(os.path.dirname(__file__), "blog")
# ---------------------------------------------------------------------------
# Shared hydration helpers
# ---------------------------------------------------------------------------
def _add_to_defpage_ctx(**kwargs: Any) -> None:
from quart import g
if not hasattr(g, '_defpage_ctx'):
g._defpage_ctx = {}
g._defpage_ctx.update(kwargs)
async def _ensure_post_data(slug: str | None) -> None:
"""Load post data and set g.post_data + defpage context.
Replicates post bp's hydrate_post_data + context_processor.
"""
from quart import g, abort
if hasattr(g, 'post_data') and g.post_data:
await _inject_post_context(g.post_data)
return
if not slug:
abort(404)
from bp.post.services.post_data import post_data
is_admin = bool((g.get("rights") or {}).get("admin"))
p_data = await post_data(slug, g.s, include_drafts=True)
if not p_data:
abort(404)
# Draft access control
if p_data["post"].get("status") != "published":
if is_admin:
pass
elif g.user and p_data["post"].get("user_id") == g.user.id:
pass
else:
abort(404)
g.post_data = p_data
g.post_slug = slug
await _inject_post_context(p_data)
async def _inject_post_context(p_data: dict) -> None:
"""Add post context_processor data to defpage context."""
from shared.config import config
from shared.infrastructure.fragments import fetch_fragment
from shared.infrastructure.data_client import fetch_data
from shared.contracts.dtos import CartSummaryDTO, dto_from_dict
from shared.infrastructure.cart_identity import current_cart_identity
db_post_id = p_data["post"]["id"]
post_slug = p_data["post"]["slug"]
container_nav = await fetch_fragment("relations", "container-nav", params={
"container_type": "page",
"container_id": str(db_post_id),
"post_slug": post_slug,
})
ctx: dict = {
**p_data,
"base_title": config()["title"],
"container_nav": container_nav,
}
if p_data["post"].get("is_page"):
ident = current_cart_identity()
summary_params: dict = {"page_slug": post_slug}
if ident["user_id"] is not None:
summary_params["user_id"] = ident["user_id"]
if ident["session_id"] is not None:
summary_params["session_id"] = ident["session_id"]
raw_summary = await fetch_data(
"cart", "cart-summary", params=summary_params, required=False,
)
page_summary = dto_from_dict(CartSummaryDTO, raw_summary) if raw_summary else CartSummaryDTO()
ctx["page_cart_count"] = (
page_summary.count + page_summary.calendar_count + page_summary.ticket_count
)
ctx["page_cart_total"] = float(
page_summary.total + page_summary.calendar_total + page_summary.ticket_total
)
_add_to_defpage_ctx(**ctx)
# ---------------------------------------------------------------------------
# Layouts
# ---------------------------------------------------------------------------
@@ -110,48 +200,48 @@ def _sub_settings_oob(ctx: dict, row_id: str, child_id: str,
def _cache_full(ctx: dict, **kw: Any) -> str:
return _sub_settings_full(ctx, "cache-row", "cache-header-child",
"settings.defpage_cache_page", "refresh", "Cache")
"defpage_cache_page", "refresh", "Cache")
def _cache_oob(ctx: dict, **kw: Any) -> str:
return _sub_settings_oob(ctx, "cache-row", "cache-header-child",
"settings.defpage_cache_page", "refresh", "Cache")
"defpage_cache_page", "refresh", "Cache")
# --- Snippets ---
def _snippets_full(ctx: dict, **kw: Any) -> str:
return _sub_settings_full(ctx, "snippets-row", "snippets-header-child",
"snippets.defpage_snippets_page", "puzzle-piece", "Snippets")
"defpage_snippets_page", "puzzle-piece", "Snippets")
def _snippets_oob(ctx: dict, **kw: Any) -> str:
return _sub_settings_oob(ctx, "snippets-row", "snippets-header-child",
"snippets.defpage_snippets_page", "puzzle-piece", "Snippets")
"defpage_snippets_page", "puzzle-piece", "Snippets")
# --- Menu Items ---
def _menu_items_full(ctx: dict, **kw: Any) -> str:
return _sub_settings_full(ctx, "menu_items-row", "menu_items-header-child",
"menu_items.defpage_menu_items_page", "bars", "Menu Items")
"defpage_menu_items_page", "bars", "Menu Items")
def _menu_items_oob(ctx: dict, **kw: Any) -> str:
return _sub_settings_oob(ctx, "menu_items-row", "menu_items-header-child",
"menu_items.defpage_menu_items_page", "bars", "Menu Items")
"defpage_menu_items_page", "bars", "Menu Items")
# --- Tag Groups ---
def _tag_groups_full(ctx: dict, **kw: Any) -> str:
return _sub_settings_full(ctx, "tag-groups-row", "tag-groups-header-child",
"blog.tag_groups_admin.defpage_tag_groups_page", "tags", "Tag Groups")
"defpage_tag_groups_page", "tags", "Tag Groups")
def _tag_groups_oob(ctx: dict, **kw: Any) -> str:
return _sub_settings_oob(ctx, "tag-groups-row", "tag-groups-header-child",
"blog.tag_groups_admin.defpage_tag_groups_page", "tags", "Tag Groups")
"defpage_tag_groups_page", "tags", "Tag Groups")
# --- Tag Group Edit ---
@@ -165,7 +255,7 @@ def _tag_group_edit_full(ctx: dict, **kw: Any) -> str:
root_hdr = root_header_sx(ctx)
settings_hdr = _settings_header_sx(ctx)
sub_hdr = _sub_settings_header_sx("tag-groups-row", "tag-groups-header-child",
qurl("blog.tag_groups_admin.defpage_tag_group_edit", id=g_id),
qurl("defpage_tag_group_edit", id=g_id),
"tags", "Tag Groups", ctx)
return "(<> " + root_hdr + " " + settings_hdr + " " + sub_hdr + ")"
@@ -178,14 +268,14 @@ def _tag_group_edit_oob(ctx: dict, **kw: Any) -> str:
from sx.sx_components import _settings_header_sx, _sub_settings_header_sx
settings_hdr_oob = _settings_header_sx(ctx, oob=True)
sub_hdr = _sub_settings_header_sx("tag-groups-row", "tag-groups-header-child",
qurl("blog.tag_groups_admin.defpage_tag_group_edit", id=g_id),
qurl("defpage_tag_group_edit", id=g_id),
"tags", "Tag Groups", ctx)
sub_oob = oob_header_sx("root-settings-header-child", "tag-groups-header-child", sub_hdr)
return "(<> " + settings_hdr_oob + " " + sub_oob + ")"
# ---------------------------------------------------------------------------
# Page helpers (sync functions available in .sx defpage expressions)
# Page helpers (async functions available in .sx defpage expressions)
# ---------------------------------------------------------------------------
def _register_blog_helpers() -> None:
@@ -208,71 +298,277 @@ def _register_blog_helpers() -> None:
})
def _h_editor_content():
# --- Editor helpers ---
async def _h_editor_content(**kw):
from sx.sx_components import render_editor_panel
return render_editor_panel()
async def _h_editor_page_content(**kw):
from sx.sx_components import render_editor_panel
return render_editor_panel(is_page=True)
# --- Post admin helpers ---
async def _h_post_admin_content(slug=None, **kw):
await _ensure_post_data(slug)
from quart import g
return getattr(g, "editor_content", "")
from sqlalchemy import select
from shared.models.page_config import PageConfig
post = (g.post_data or {}).get("post", {})
features = {}
sumup_configured = False
sumup_merchant_code = ""
sumup_checkout_prefix = ""
if post.get("is_page"):
pc = (await g.s.execute(
select(PageConfig).where(
PageConfig.container_type == "page",
PageConfig.container_id == post["id"],
)
)).scalar_one_or_none()
if pc:
features = pc.features or {}
sumup_configured = bool(pc.sumup_api_key)
sumup_merchant_code = pc.sumup_merchant_code or ""
sumup_checkout_prefix = pc.sumup_checkout_prefix or ""
from shared.sx.page import get_template_context
from sx.sx_components import _post_admin_main_panel_sx
tctx = await get_template_context()
tctx.update({
"features": features,
"sumup_configured": sumup_configured,
"sumup_merchant_code": sumup_merchant_code,
"sumup_checkout_prefix": sumup_checkout_prefix,
})
return _post_admin_main_panel_sx(tctx)
def _h_editor_page_content():
async def _h_post_data_content(slug=None, **kw):
await _ensure_post_data(slug)
from shared.sx.page import get_template_context
from sx.sx_components import _post_data_content_sx
tctx = await get_template_context()
return _post_data_content_sx(tctx)
async def _h_post_preview_content(slug=None, **kw):
await _ensure_post_data(slug)
from quart import g
return getattr(g, "editor_page_content", "")
from models.ghost_content import Post
from sqlalchemy import select as sa_select
post_id = g.post_data["post"]["id"]
post = (await g.s.execute(
sa_select(Post).where(Post.id == post_id)
)).scalar_one_or_none()
preview_ctx: dict = {}
sx_content = getattr(post, "sx_content", None) or ""
if sx_content:
from shared.sx.prettify import sx_to_pretty_sx
preview_ctx["sx_pretty"] = sx_to_pretty_sx(sx_content)
lexical_raw = getattr(post, "lexical", None) or ""
if lexical_raw:
from shared.sx.prettify import json_to_pretty_sx
preview_ctx["json_pretty"] = json_to_pretty_sx(lexical_raw)
if sx_content:
from shared.sx.parser import parse as sx_parse
from shared.sx.html import render as sx_html_render
from shared.sx.jinja_bridge import _COMPONENT_ENV
try:
parsed = sx_parse(sx_content)
preview_ctx["sx_rendered"] = sx_html_render(parsed, dict(_COMPONENT_ENV))
except Exception:
preview_ctx["sx_rendered"] = "<em>Error rendering sx</em>"
if lexical_raw:
from bp.blog.ghost.lexical_renderer import render_lexical
try:
preview_ctx["lex_rendered"] = render_lexical(lexical_raw)
except Exception:
preview_ctx["lex_rendered"] = "<em>Error rendering lexical</em>"
from shared.sx.page import get_template_context
from sx.sx_components import _preview_main_panel_sx
tctx = await get_template_context()
tctx.update(preview_ctx)
return _preview_main_panel_sx(tctx)
def _h_post_admin_content():
async def _h_post_entries_content(slug=None, **kw):
await _ensure_post_data(slug)
from quart import g
return getattr(g, "post_admin_content", "")
from sqlalchemy import select
from shared.models.calendars import Calendar
from bp.post.services.entry_associations import get_post_entry_ids
post_id = g.post_data["post"]["id"]
associated_entry_ids = await get_post_entry_ids(post_id)
result = await g.s.execute(
select(Calendar)
.where(Calendar.deleted_at.is_(None))
.order_by(Calendar.name.asc())
)
all_calendars = result.scalars().all()
for calendar in all_calendars:
await g.s.refresh(calendar, ["entries", "post"])
from shared.sx.page import get_template_context
from sx.sx_components import _post_entries_content_sx
tctx = await get_template_context()
tctx["all_calendars"] = all_calendars
tctx["associated_entry_ids"] = associated_entry_ids
return _post_entries_content_sx(tctx)
def _h_post_data_content():
async def _h_post_settings_content(slug=None, **kw):
await _ensure_post_data(slug)
from quart import g, request
from models.ghost_content import Post
from sqlalchemy import select as sa_select
from sqlalchemy.orm import selectinload
from bp.post.admin.routes import _post_to_edit_dict
post_id = g.post_data["post"]["id"]
post = (await g.s.execute(
sa_select(Post)
.where(Post.id == post_id)
.options(selectinload(Post.tags))
)).scalar_one_or_none()
ghost_post = _post_to_edit_dict(post) if post else {}
save_success = request.args.get("saved") == "1"
from shared.sx.page import get_template_context
from sx.sx_components import _post_settings_content_sx
tctx = await get_template_context()
tctx["ghost_post"] = ghost_post
tctx["save_success"] = save_success
return _post_settings_content_sx(tctx)
async def _h_post_edit_content(slug=None, **kw):
await _ensure_post_data(slug)
from quart import g, request
from models.ghost_content import Post
from sqlalchemy import select as sa_select
from sqlalchemy.orm import selectinload
from shared.infrastructure.data_client import fetch_data
from bp.post.admin.routes import _post_to_edit_dict
post_id = g.post_data["post"]["id"]
post = (await g.s.execute(
sa_select(Post)
.where(Post.id == post_id)
.options(selectinload(Post.tags))
)).scalar_one_or_none()
ghost_post = _post_to_edit_dict(post) if post else {}
save_success = request.args.get("saved") == "1"
save_error = request.args.get("error", "")
raw_newsletters = await fetch_data("account", "newsletters", required=False) or []
from types import SimpleNamespace
newsletters = [SimpleNamespace(**nl) for nl in raw_newsletters]
from shared.sx.page import get_template_context
from sx.sx_components import _post_edit_content_sx
tctx = await get_template_context()
tctx["ghost_post"] = ghost_post
tctx["save_success"] = save_success
tctx["save_error"] = save_error
tctx["newsletters"] = newsletters
return _post_edit_content_sx(tctx)
# --- Settings helpers ---
async def _h_settings_content(**kw):
from shared.sx.page import get_template_context
from sx.sx_components import _settings_main_panel_sx
tctx = await get_template_context()
return _settings_main_panel_sx(tctx)
async def _h_cache_content(**kw):
from shared.sx.page import get_template_context
from sx.sx_components import _cache_main_panel_sx
tctx = await get_template_context()
return _cache_main_panel_sx(tctx)
# --- Snippets helper ---
async def _h_snippets_content(**kw):
from quart import g
return getattr(g, "post_data_content", "")
from sqlalchemy import select, or_
from models import Snippet
uid = g.user.id
is_admin = g.rights.get("admin")
filters = [Snippet.user_id == uid, Snippet.visibility == "shared"]
if is_admin:
filters.append(Snippet.visibility == "admin")
rows = (await g.s.execute(
select(Snippet).where(or_(*filters)).order_by(Snippet.name)
)).scalars().all()
from shared.sx.page import get_template_context
from sx.sx_components import _snippets_main_panel_sx
tctx = await get_template_context()
tctx["snippets"] = rows
tctx["is_admin"] = is_admin
return _snippets_main_panel_sx(tctx)
def _h_post_preview_content():
# --- Menu Items helper ---
async def _h_menu_items_content(**kw):
from quart import g
return getattr(g, "post_preview_content", "")
from bp.menu_items.services.menu_items import get_all_menu_items
menu_items = await get_all_menu_items(g.s)
from shared.sx.page import get_template_context
from sx.sx_components import _menu_items_main_panel_sx
tctx = await get_template_context()
tctx["menu_items"] = menu_items
return _menu_items_main_panel_sx(tctx)
def _h_post_entries_content():
# --- Tag Groups helpers ---
async def _h_tag_groups_content(**kw):
from quart import g
return getattr(g, "post_entries_content", "")
from sqlalchemy import select
from models.tag_group import TagGroup
from bp.blog.admin.routes import _unassigned_tags
groups = list(
(await g.s.execute(
select(TagGroup).order_by(TagGroup.sort_order, TagGroup.name)
)).scalars()
)
unassigned = await _unassigned_tags(g.s)
from shared.sx.page import get_template_context
from sx.sx_components import _tag_groups_main_panel_sx
tctx = await get_template_context()
tctx.update({"groups": groups, "unassigned_tags": unassigned})
return _tag_groups_main_panel_sx(tctx)
def _h_post_settings_content():
from quart import g
return getattr(g, "post_settings_content", "")
def _h_post_edit_content():
from quart import g
return getattr(g, "post_edit_content", "")
def _h_settings_content():
from quart import g
return getattr(g, "settings_content", "")
def _h_cache_content():
from quart import g
return getattr(g, "cache_content", "")
def _h_snippets_content():
from quart import g
return getattr(g, "snippets_content", "")
def _h_menu_items_content():
from quart import g
return getattr(g, "menu_items_content", "")
def _h_tag_groups_content():
from quart import g
return getattr(g, "tag_groups_content", "")
def _h_tag_group_edit_content():
from quart import g
return getattr(g, "tag_group_edit_content", "")
async def _h_tag_group_edit_content(id=None, **kw):
from quart import g, abort
from sqlalchemy import select
from models.tag_group import TagGroup, TagGroupTag
from models.ghost_content import Tag
tg = await g.s.get(TagGroup, id)
if not tg:
abort(404)
assigned_rows = list(
(await g.s.execute(
select(TagGroupTag.tag_id).where(TagGroupTag.tag_group_id == id)
)).scalars()
)
all_tags = list(
(await g.s.execute(
select(Tag).where(
Tag.deleted_at.is_(None),
(Tag.visibility == "public") | (Tag.visibility.is_(None)),
).order_by(Tag.name)
)).scalars()
)
from shared.sx.page import get_template_context
from sx.sx_components import _tag_groups_edit_main_panel_sx
tctx = await get_template_context()
tctx.update({
"group": tg,
"all_tags": all_tags,
"assigned_tag_ids": set(assigned_rows),
})
return _tag_groups_edit_main_panel_sx(tctx)

View File

@@ -15,54 +15,54 @@
:layout :blog
:content (editor-page-content))
; --- Post admin pages (nested under /<slug>/admin/) ---
; --- Post admin pages (absolute paths under /<slug>/admin/) ---
(defpage post-admin
:path "/"
:path "/<slug>/admin/"
:auth :admin
:layout (:post-admin :selected "admin")
:content (post-admin-content))
:content (post-admin-content slug))
(defpage post-data
:path "/data/"
:path "/<slug>/admin/data/"
:auth :admin
:layout (:post-admin :selected "data")
:content (post-data-content))
:content (post-data-content slug))
(defpage post-preview
:path "/preview/"
:path "/<slug>/admin/preview/"
:auth :admin
:layout (:post-admin :selected "preview")
:content (post-preview-content))
:content (post-preview-content slug))
(defpage post-entries
:path "/entries/"
:path "/<slug>/admin/entries/"
:auth :admin
:layout (:post-admin :selected "entries")
:content (post-entries-content))
:content (post-entries-content slug))
(defpage post-settings
:path "/settings/"
:path "/<slug>/admin/settings/"
:auth :post_author
:layout (:post-admin :selected "settings")
:content (post-settings-content))
:content (post-settings-content slug))
(defpage post-edit
:path "/edit/"
:path "/<slug>/admin/edit/"
:auth :post_author
:layout (:post-admin :selected "edit")
:content (post-edit-content))
:content (post-edit-content slug))
; --- Settings pages ---
; --- Settings pages (absolute paths) ---
(defpage settings-home
:path "/"
:path "/settings/"
:auth :admin
:layout :blog-settings
:content (settings-content))
(defpage cache-page
:path "/cache/"
:path "/settings/cache/"
:auth :admin
:layout :blog-cache
:content (cache-content))
@@ -70,7 +70,7 @@
; --- Snippets ---
(defpage snippets-page
:path "/"
:path "/settings/snippets/"
:auth :login
:layout :blog-snippets
:content (snippets-content))
@@ -78,7 +78,7 @@
; --- Menu Items ---
(defpage menu-items-page
:path "/"
:path "/settings/menu_items/"
:auth :admin
:layout :blog-menu-items
:content (menu-items-content))
@@ -86,13 +86,13 @@
; --- Tag Groups ---
(defpage tag-groups-page
:path "/"
:path "/settings/tag-groups/"
:auth :admin
:layout :blog-tag-groups
:content (tag-groups-content))
(defpage tag-group-edit
:path "/<int:id>/"
:path "/settings/tag-groups/<int:id>/"
:auth :admin
:layout :blog-tag-group-edit
:content (tag-group-edit-content))
:content (tag-group-edit-content id))