giles
f0fbcef3f6
Move _blog_header_sx, _settings_header_sx, _settings_nav_sx, and _sub_settings_header_sx into the layout module as local helpers. Eliminates 14 imports from sx_components.py for the layout system. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
462 lines
17 KiB
Python
462 lines
17 KiB
Python
"""Blog defpage setup — registers layouts, page helpers, and loads .sx pages."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
def setup_blog_pages() -> None:
|
|
"""Register blog-specific layouts, page helpers, and load page definitions."""
|
|
_register_blog_layouts()
|
|
_register_blog_helpers()
|
|
_load_blog_page_files()
|
|
|
|
|
|
def _load_blog_page_files() -> None:
|
|
import os
|
|
from shared.sx.pages import load_page_dir
|
|
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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Header helpers (moved from sx_components — thin render_to_sx wrappers)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _blog_header_sx(ctx: dict, *, oob: bool = False) -> str:
|
|
from shared.sx.helpers import render_to_sx
|
|
from shared.sx.parser import SxExpr
|
|
return await render_to_sx("menu-row-sx",
|
|
id="blog-row", level=1,
|
|
link_label_content=SxExpr("(div)"),
|
|
child_id="blog-header-child", oob=oob)
|
|
|
|
|
|
async def _settings_header_sx(ctx: dict, *, oob: bool = False) -> str:
|
|
from shared.sx.helpers import render_to_sx
|
|
from shared.sx.parser import SxExpr
|
|
from quart import url_for as qurl
|
|
|
|
settings_href = qurl("settings.defpage_settings_home")
|
|
label_sx = await render_to_sx("blog-admin-label")
|
|
nav_sx = await _settings_nav_sx(ctx)
|
|
|
|
return await render_to_sx("menu-row-sx",
|
|
id="root-settings-row", level=1,
|
|
link_href=settings_href,
|
|
link_label_content=SxExpr(label_sx),
|
|
nav=SxExpr(nav_sx) if nav_sx else None,
|
|
child_id="root-settings-header-child", oob=oob)
|
|
|
|
|
|
async def _settings_nav_sx(ctx: dict) -> str:
|
|
from shared.sx.helpers import render_to_sx
|
|
from quart import url_for as qurl
|
|
|
|
select_colours = ctx.get("select_colours", "")
|
|
parts = []
|
|
for endpoint, icon, label in [
|
|
("menu_items.defpage_menu_items_page", "bars", "Menu Items"),
|
|
("snippets.defpage_snippets_page", "puzzle-piece", "Snippets"),
|
|
("blog.tag_groups_admin.defpage_tag_groups_page", "tags", "Tag Groups"),
|
|
("settings.defpage_cache_page", "refresh", "Cache"),
|
|
]:
|
|
href = qurl(endpoint)
|
|
parts.append(await render_to_sx("nav-link",
|
|
href=href, icon=f"fa fa-{icon}", label=label,
|
|
select_colours=select_colours))
|
|
return "(<> " + " ".join(parts) + ")" if parts else ""
|
|
|
|
|
|
async def _sub_settings_header_sx(row_id: str, child_id: str, href: str,
|
|
icon: str, label: str, ctx: dict,
|
|
*, oob: bool = False, nav_sx: str = "") -> str:
|
|
from shared.sx.helpers import render_to_sx
|
|
from shared.sx.parser import SxExpr
|
|
|
|
label_sx = await render_to_sx("blog-sub-settings-label",
|
|
icon=f"fa fa-{icon}", label=label)
|
|
return await render_to_sx("menu-row-sx",
|
|
id=row_id, level=2,
|
|
link_href=href,
|
|
link_label_content=SxExpr(label_sx),
|
|
nav=SxExpr(nav_sx) if nav_sx else None,
|
|
child_id=child_id, oob=oob)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Layouts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _register_blog_layouts() -> None:
|
|
from shared.sx.layouts import register_custom_layout
|
|
register_custom_layout("blog", _blog_full, _blog_oob)
|
|
register_custom_layout("blog-settings", _settings_full, _settings_oob,
|
|
mobile_fn=_settings_mobile)
|
|
register_custom_layout("blog-cache", _cache_full, _cache_oob)
|
|
register_custom_layout("blog-snippets", _snippets_full, _snippets_oob)
|
|
register_custom_layout("blog-menu-items", _menu_items_full, _menu_items_oob)
|
|
register_custom_layout("blog-tag-groups", _tag_groups_full, _tag_groups_oob)
|
|
register_custom_layout("blog-tag-group-edit",
|
|
_tag_group_edit_full, _tag_group_edit_oob)
|
|
|
|
|
|
# --- Blog layout (root + blog header) ---
|
|
|
|
async def _blog_full(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx
|
|
root_hdr = await root_header_sx(ctx)
|
|
blog_hdr = await _blog_header_sx(ctx)
|
|
return "(<> " + root_hdr + " " + blog_hdr + ")"
|
|
|
|
|
|
async def _blog_oob(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx, oob_header_sx
|
|
root_hdr = await root_header_sx(ctx)
|
|
blog_hdr = await _blog_header_sx(ctx)
|
|
rows = "(<> " + root_hdr + " " + blog_hdr + ")"
|
|
return await oob_header_sx("root-header-child", "blog-header-child", rows)
|
|
|
|
|
|
# --- Settings layout (root + settings header) ---
|
|
|
|
async def _settings_full(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx
|
|
root_hdr = await root_header_sx(ctx)
|
|
settings_hdr = await _settings_header_sx(ctx)
|
|
return "(<> " + root_hdr + " " + settings_hdr + ")"
|
|
|
|
|
|
async def _settings_oob(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx, oob_header_sx
|
|
root_hdr = await root_header_sx(ctx)
|
|
settings_hdr = await _settings_header_sx(ctx)
|
|
rows = "(<> " + root_hdr + " " + settings_hdr + ")"
|
|
return await oob_header_sx("root-header-child", "root-settings-header-child", rows)
|
|
|
|
|
|
async def _settings_mobile(ctx: dict, **kw: Any) -> str:
|
|
return await _settings_nav_sx(ctx)
|
|
|
|
|
|
# --- Sub-settings helpers ---
|
|
|
|
async def _sub_settings_full(ctx: dict, row_id: str, child_id: str,
|
|
endpoint: str, icon: str, label: str) -> str:
|
|
from shared.sx.helpers import root_header_sx
|
|
from quart import url_for as qurl
|
|
root_hdr = await root_header_sx(ctx)
|
|
settings_hdr = await _settings_header_sx(ctx)
|
|
sub_hdr = await _sub_settings_header_sx(row_id, child_id,
|
|
qurl(endpoint), icon, label, ctx)
|
|
return "(<> " + root_hdr + " " + settings_hdr + " " + sub_hdr + ")"
|
|
|
|
|
|
async def _sub_settings_oob(ctx: dict, row_id: str, child_id: str,
|
|
endpoint: str, icon: str, label: str) -> str:
|
|
from shared.sx.helpers import oob_header_sx
|
|
from quart import url_for as qurl
|
|
settings_hdr_oob = await _settings_header_sx(ctx, oob=True)
|
|
sub_hdr = await _sub_settings_header_sx(row_id, child_id,
|
|
qurl(endpoint), icon, label, ctx)
|
|
sub_oob = await oob_header_sx("root-settings-header-child", child_id, sub_hdr)
|
|
return "(<> " + settings_hdr_oob + " " + sub_oob + ")"
|
|
|
|
|
|
# --- Cache ---
|
|
|
|
async def _cache_full(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_full(ctx, "cache-row", "cache-header-child",
|
|
"defpage_cache_page", "refresh", "Cache")
|
|
|
|
|
|
async def _cache_oob(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_oob(ctx, "cache-row", "cache-header-child",
|
|
"defpage_cache_page", "refresh", "Cache")
|
|
|
|
|
|
# --- Snippets ---
|
|
|
|
async def _snippets_full(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_full(ctx, "snippets-row", "snippets-header-child",
|
|
"defpage_snippets_page", "puzzle-piece", "Snippets")
|
|
|
|
|
|
async def _snippets_oob(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_oob(ctx, "snippets-row", "snippets-header-child",
|
|
"defpage_snippets_page", "puzzle-piece", "Snippets")
|
|
|
|
|
|
# --- Menu Items ---
|
|
|
|
async def _menu_items_full(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_full(ctx, "menu_items-row", "menu_items-header-child",
|
|
"defpage_menu_items_page", "bars", "Menu Items")
|
|
|
|
|
|
async def _menu_items_oob(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_oob(ctx, "menu_items-row", "menu_items-header-child",
|
|
"defpage_menu_items_page", "bars", "Menu Items")
|
|
|
|
|
|
# --- Tag Groups ---
|
|
|
|
async def _tag_groups_full(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_full(ctx, "tag-groups-row", "tag-groups-header-child",
|
|
"defpage_tag_groups_page", "tags", "Tag Groups")
|
|
|
|
|
|
async def _tag_groups_oob(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_oob(ctx, "tag-groups-row", "tag-groups-header-child",
|
|
"defpage_tag_groups_page", "tags", "Tag Groups")
|
|
|
|
|
|
# --- Tag Group Edit ---
|
|
|
|
async def _tag_group_edit_full(ctx: dict, **kw: Any) -> str:
|
|
from quart import request, url_for as qurl
|
|
from shared.sx.helpers import root_header_sx
|
|
g_id = (request.view_args or {}).get("id")
|
|
root_hdr = await root_header_sx(ctx)
|
|
settings_hdr = await _settings_header_sx(ctx)
|
|
sub_hdr = await _sub_settings_header_sx("tag-groups-row", "tag-groups-header-child",
|
|
qurl("defpage_tag_group_edit", id=g_id),
|
|
"tags", "Tag Groups", ctx)
|
|
return "(<> " + root_hdr + " " + settings_hdr + " " + sub_hdr + ")"
|
|
|
|
|
|
async def _tag_group_edit_oob(ctx: dict, **kw: Any) -> str:
|
|
from quart import request, url_for as qurl
|
|
from shared.sx.helpers import oob_header_sx
|
|
g_id = (request.view_args or {}).get("id")
|
|
settings_hdr_oob = await _settings_header_sx(ctx, oob=True)
|
|
sub_hdr = await _sub_settings_header_sx("tag-groups-row", "tag-groups-header-child",
|
|
qurl("defpage_tag_group_edit", id=g_id),
|
|
"tags", "Tag Groups", ctx)
|
|
sub_oob = await oob_header_sx("root-settings-header-child", "tag-groups-header-child", sub_hdr)
|
|
return "(<> " + settings_hdr_oob + " " + sub_oob + ")"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Page helpers (async functions available in .sx defpage expressions)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _register_blog_helpers() -> None:
|
|
from shared.sx.pages import register_page_helpers
|
|
register_page_helpers("blog", {
|
|
"editor-content": _h_editor_content,
|
|
"editor-page-content": _h_editor_page_content,
|
|
"post-admin-content": _h_post_admin_content,
|
|
"post-data-content": _h_post_data_content,
|
|
"post-preview-content": _h_post_preview_content,
|
|
"post-entries-content": _h_post_entries_content,
|
|
"post-settings-content": _h_post_settings_content,
|
|
"post-edit-content": _h_post_edit_content,
|
|
})
|
|
|
|
|
|
# --- Editor helpers ---
|
|
|
|
async def _h_editor_content(**kw):
|
|
from sx.sx_components import render_editor_panel
|
|
return await render_editor_panel()
|
|
|
|
|
|
async def _h_editor_page_content(**kw):
|
|
from sx.sx_components import render_editor_panel
|
|
return await render_editor_panel(is_page=True)
|
|
|
|
|
|
# --- Post admin helpers ---
|
|
|
|
async def _h_post_admin_content(slug=None, **kw):
|
|
await _ensure_post_data(slug)
|
|
return '(div :class "pb-8")'
|
|
|
|
|
|
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
|
|
from shared.services.registry import services
|
|
from shared.sx.page import get_template_context
|
|
from sx.sx_components import _preview_main_panel_sx
|
|
preview_data = await services.get("blog_page").preview_data(g.s)
|
|
tctx = await get_template_context()
|
|
tctx.update(preview_data)
|
|
return await _preview_main_panel_sx(tctx)
|
|
|
|
|
|
async def _h_post_entries_content(slug=None, **kw):
|
|
await _ensure_post_data(slug)
|
|
from quart import g
|
|
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 await _post_entries_content_sx(tctx)
|
|
|
|
|
|
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 await _post_edit_content_sx(tctx)
|
|
|
|
|