giles
959e63d440
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 1m44s
Replace ~250 render_to_sx calls across all services with sync sx_call, converting many async functions to sync where no other awaits remained. Make render_to_sx/render_to_sx_with_env private (_render_to_sx). Add (post-header-ctx) IO primitive and shared post/post-admin defmacros. Convert built-in post/post-admin layouts from Python to register_sx_layout with .sx defcomps. Remove dead post_admin_mobile_nav_sx. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
307 lines
12 KiB
Python
307 lines
12 KiB
Python
from werkzeug.exceptions import HTTPException
|
|
from shared.utils import hx_fragment_request
|
|
|
|
from quart import (
|
|
request,
|
|
render_template,
|
|
make_response,
|
|
current_app
|
|
)
|
|
|
|
from markupsafe import escape
|
|
from shared.infrastructure.fragments import FragmentError as _FragmentError
|
|
|
|
class AppError(ValueError):
|
|
"""
|
|
Base class for app-level, client-safe errors.
|
|
Behaves like ValueError so existing except ValueError: still works.
|
|
"""
|
|
status_code: int = 400
|
|
|
|
def __init__(self, message, *, status_code: int | None = None):
|
|
# Support a single message or a list/tuple of messages
|
|
if isinstance(message, (list, tuple, set)):
|
|
self.messages = [str(m) for m in message]
|
|
msg = self.messages[0] if self.messages else ""
|
|
else:
|
|
self.messages = [str(message)]
|
|
msg = str(message)
|
|
|
|
super().__init__(msg)
|
|
|
|
if status_code is not None:
|
|
self.status_code = status_code
|
|
|
|
|
|
def _error_page(message: str) -> str:
|
|
"""Self-contained error page HTML. Bypasses Jinja/context processors."""
|
|
return (
|
|
"<!DOCTYPE html><html><head><meta charset='utf-8'>"
|
|
"<meta name='viewport' content='width=device-width,initial-scale=1'>"
|
|
"<title>Error</title>"
|
|
"<style>"
|
|
"body{margin:0;min-height:100vh;display:flex;align-items:center;"
|
|
"justify-content:center;font-family:system-ui,sans-serif;"
|
|
"background:#fafafa;color:#333}"
|
|
".box{text-align:center;padding:2rem;max-width:480px}"
|
|
".box h1{font-size:1.5rem;color:#ef4444;margin:0 0 1rem}"
|
|
".box p{margin:0 0 1.5rem;line-height:1.6}"
|
|
".box a{color:#3b82f6;text-decoration:none}"
|
|
".box img{max-width:300px;margin:1rem auto}"
|
|
"</style></head><body>"
|
|
"<div class='box'>"
|
|
f"<h1>{message}</h1>"
|
|
"<img src='/static/errors/error.gif' width='300' height='300'>"
|
|
"<p><a href='javascript:location.reload()'>Reload</a></p>"
|
|
"</div></body></html>"
|
|
)
|
|
|
|
|
|
def _sx_error_page(errnum: str, message: str, image: str | None = None) -> str:
|
|
"""Render an error page via s-expressions. Bypasses Jinja entirely."""
|
|
from shared.sx.page import render_page
|
|
|
|
return render_page(
|
|
'(~error-page :title title :message message :image image :asset-url "/static")',
|
|
title=f"{errnum} Error",
|
|
message=message,
|
|
image=image,
|
|
)
|
|
|
|
|
|
async def _rich_error_page(errnum: str, message: str, image: str | None = None) -> str | None:
|
|
"""Try to render an error page with site headers and post breadcrumb.
|
|
|
|
Returns HTML string on success, None if we should fall back to the
|
|
minimal error page. All failures are swallowed — this must never
|
|
make a bad situation worse.
|
|
"""
|
|
from quart import g
|
|
|
|
# Skip for internal/static requests
|
|
if request.path.startswith(("/internal/", "/static/", "/auth/")):
|
|
return None
|
|
|
|
try:
|
|
# Build a minimal context — avoid get_template_context() which
|
|
# calls cross-service fragment fetches that may fail.
|
|
from shared.infrastructure.context import base_context
|
|
try:
|
|
ctx = await base_context()
|
|
except Exception:
|
|
ctx = {"base_title": "Rose Ash", "asset_url": "/static"}
|
|
|
|
# Inject Jinja globals (blog_url, cart_url, etc.) — these are
|
|
# needed by call_url() for cross-subdomain links.
|
|
for key, val in current_app.jinja_env.globals.items():
|
|
if key not in ctx:
|
|
ctx[key] = val
|
|
|
|
# Try to fetch fragments, but don't fail if they're unreachable
|
|
try:
|
|
from shared.infrastructure.fragments import fetch_fragments
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
user = getattr(g, "user", None)
|
|
ident = current_cart_identity()
|
|
cart_params = {}
|
|
if ident["user_id"] is not None:
|
|
cart_params["user_id"] = ident["user_id"]
|
|
if ident["session_id"] is not None:
|
|
cart_params["session_id"] = ident["session_id"]
|
|
cart_mini, auth_menu, nav_tree = await fetch_fragments([
|
|
("cart", "cart-mini", cart_params or None),
|
|
("account", "auth-menu", {"email": user.email} if user else None),
|
|
("blog", "nav-tree", {"app_name": current_app.name, "path": request.path}),
|
|
])
|
|
ctx["cart_mini"] = cart_mini
|
|
ctx["auth_menu"] = auth_menu
|
|
ctx["nav_tree"] = nav_tree
|
|
except Exception:
|
|
ctx.setdefault("cart_mini", "")
|
|
ctx.setdefault("auth_menu", "")
|
|
ctx.setdefault("nav_tree", "")
|
|
|
|
# Try to hydrate post data from slug if not already available
|
|
segments = [s for s in request.path.strip("/").split("/") if s]
|
|
slug = segments[0] if segments else None
|
|
post_data = getattr(g, "post_data", None)
|
|
|
|
if slug and not post_data:
|
|
try:
|
|
from shared.infrastructure.data_client import fetch_data
|
|
raw = await fetch_data(
|
|
"blog", "post-by-slug",
|
|
params={"slug": slug},
|
|
required=False,
|
|
)
|
|
if raw:
|
|
post_data = {
|
|
"post": {
|
|
"id": raw["id"],
|
|
"title": raw["title"],
|
|
"slug": raw["slug"],
|
|
"feature_image": raw.get("feature_image"),
|
|
},
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
# Root header (site nav bar)
|
|
from shared.sx.helpers import (
|
|
root_header_sx, post_header_sx,
|
|
header_child_sx, full_page_sx, sx_call,
|
|
)
|
|
hdr = await root_header_sx(ctx)
|
|
|
|
# Post breadcrumb if we resolved a post
|
|
post = (post_data or {}).get("post") or ctx.get("post") or {}
|
|
if post.get("slug"):
|
|
ctx["post"] = post
|
|
post_row = await post_header_sx(ctx)
|
|
if post_row:
|
|
hdr = "(<> " + hdr + " " + await header_child_sx(post_row) + ")"
|
|
|
|
# Error content
|
|
error_content = sx_call("error-content", errnum=errnum, message=message, image=image)
|
|
|
|
return await full_page_sx(ctx, header_rows=hdr, content=error_content)
|
|
except Exception:
|
|
current_app.logger.debug("Rich error page failed, falling back", exc_info=True)
|
|
return None
|
|
|
|
|
|
def errors(app):
|
|
def _info(e):
|
|
return {
|
|
"exception": e,
|
|
"method": request.method,
|
|
"url": str(request.url),
|
|
"base_url": str(request.base_url),
|
|
"root_path": request.root_path,
|
|
"path": request.path,
|
|
"full_path": request.full_path,
|
|
"endpoint": request.endpoint,
|
|
"url_rule": str(request.url_rule) if request.url_rule else None,
|
|
"headers": {k: v for k, v in request.headers.items()
|
|
if k.lower().startswith("x-forwarded") or k in ("Host",)},
|
|
}
|
|
|
|
@app.errorhandler(404)
|
|
async def not_found(e):
|
|
current_app.logger.warning("404 %s", _info(e))
|
|
if hx_fragment_request():
|
|
html = await render_template(
|
|
"_types/root/exceptions/hx/_.html",
|
|
errnum='404'
|
|
)
|
|
else:
|
|
# Try a rich error page with site headers + post breadcrumb
|
|
html = await _rich_error_page(
|
|
"404", "NOT FOUND",
|
|
image="/static/errors/404.gif",
|
|
)
|
|
if html is None:
|
|
try:
|
|
html = _sx_error_page(
|
|
"404", "NOT FOUND",
|
|
image="/static/errors/404.gif",
|
|
)
|
|
except Exception:
|
|
html = await render_template(
|
|
"_types/root/exceptions/_.html",
|
|
errnum='404',
|
|
)
|
|
|
|
return await make_response(html, 404)
|
|
|
|
@app.errorhandler(403)
|
|
async def not_allowed(e):
|
|
current_app.logger.warning("403 %s", _info(e))
|
|
if hx_fragment_request():
|
|
html = await render_template(
|
|
"_types/root/exceptions/hx/_.html",
|
|
errnum='403'
|
|
)
|
|
else:
|
|
html = await _rich_error_page(
|
|
"403", "FORBIDDEN",
|
|
image="/static/errors/403.gif",
|
|
)
|
|
if html is None:
|
|
try:
|
|
html = _sx_error_page(
|
|
"403", "FORBIDDEN",
|
|
image="/static/errors/403.gif",
|
|
)
|
|
except Exception:
|
|
html = await render_template(
|
|
"_types/root/exceptions/_.html",
|
|
errnum='403',
|
|
)
|
|
|
|
return await make_response(html, 403)
|
|
|
|
@app.errorhandler(AppError)
|
|
async def app_error(e: AppError):
|
|
# App-level, client-safe errors
|
|
current_app.logger.info("AppError %s", _info(e))
|
|
status = getattr(e, "status_code", 400)
|
|
messages = getattr(e, "messages", [str(e)])
|
|
|
|
if request.headers.get("SX-Request") == "true" or request.headers.get("HX-Request") == "true":
|
|
from shared.sx.jinja_bridge import render as render_comp
|
|
items = "".join(
|
|
render_comp("error-list-item", message=str(escape(m)))
|
|
for m in messages if m
|
|
)
|
|
html = render_comp("error-list", items_html=items)
|
|
return await make_response(html, status)
|
|
|
|
# Non-HTMX: show a nicer page with error messages
|
|
html = await render_template(
|
|
"_types/root/exceptions/app_error.html",
|
|
messages=messages,
|
|
)
|
|
return await make_response(html, status)
|
|
|
|
@app.errorhandler(_FragmentError)
|
|
async def fragment_error(e):
|
|
current_app.logger.error("FragmentError %s", _info(e))
|
|
msg = str(e)
|
|
# Extract service name from "Fragment account/auth-menu failed: ..."
|
|
service = msg.split("/")[0].replace("Fragment ", "") if "/" in msg else "unknown"
|
|
if request.headers.get("SX-Request") == "true" or request.headers.get("HX-Request") == "true":
|
|
from shared.sx.jinja_bridge import render as render_comp
|
|
return await make_response(
|
|
render_comp("fragment-error", service=str(escape(service))),
|
|
503,
|
|
)
|
|
# Raw HTML — cannot use render_template here because the context
|
|
# processor would try to fetch fragments again → infinite loop.
|
|
return await make_response(_error_page(
|
|
f"The <b>{escape(service)}</b> service is currently unavailable. It may be restarting."
|
|
), 503)
|
|
|
|
@app.errorhandler(Exception)
|
|
async def error(e):
|
|
current_app.logger.exception("Exception %s", _info(e))
|
|
|
|
status = 500
|
|
if isinstance(e, HTTPException):
|
|
status = e.code or 500
|
|
|
|
if request.headers.get("SX-Request") == "true" or request.headers.get("HX-Request") == "true":
|
|
return await make_response(
|
|
"Something went wrong. Please try again.",
|
|
status,
|
|
)
|
|
|
|
errnum = str(status)
|
|
html = await _rich_error_page(
|
|
errnum, "WELL THIS IS EMBARRASSING\u2026",
|
|
image="/static/errors/error.gif",
|
|
)
|
|
if html is None:
|
|
html = _error_page("WELL THIS IS EMBARRASSING…")
|
|
return await make_response(html, status)
|