giles
f1b7fdd37d
Build a minimal context directly instead of relying on get_template_context() which runs the full context processor chain including cross-service fragment fetches. Each step (base_context, fragments, post hydration) is independently try/excepted so the page renders with whatever is available. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
317 lines
12 KiB
Python
317 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 _sexp_error_page(errnum: str, message: str, image: str | None = None) -> str:
|
|
"""Render an error page via s-expressions. Bypasses Jinja entirely."""
|
|
from shared.sexp.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:
|
|
from shared.sexp.jinja_bridge import render
|
|
from shared.sexp.helpers import full_page, call_url
|
|
|
|
# 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"}
|
|
|
|
# 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_html, auth_menu_html, nav_tree_html = 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_html"] = cart_mini_html
|
|
ctx["auth_menu_html"] = auth_menu_html
|
|
ctx["nav_tree_html"] = nav_tree_html
|
|
except Exception:
|
|
ctx.setdefault("cart_mini_html", "")
|
|
ctx.setdefault("auth_menu_html", "")
|
|
ctx.setdefault("nav_tree_html", "")
|
|
|
|
# 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.sexp.helpers import root_header_html
|
|
hdr = root_header_html(ctx)
|
|
|
|
# Post breadcrumb if we resolved a post
|
|
post = (post_data or {}).get("post") or ctx.get("post") or {}
|
|
if post.get("slug"):
|
|
title = (post.get("title") or "")[:160]
|
|
feature_image = post.get("feature_image")
|
|
label_html = ""
|
|
if feature_image:
|
|
label_html += (
|
|
f'<img src="{feature_image}" '
|
|
f'class="h-8 w-8 rounded-full object-cover border border-stone-300 flex-shrink-0">'
|
|
)
|
|
label_html += f"<span>{escape(title)}</span>"
|
|
post_row = render(
|
|
"menu-row",
|
|
id="post-row", level=1,
|
|
link_href=call_url(ctx, "blog_url", f"/{post['slug']}/"),
|
|
link_label_html=label_html,
|
|
child_id="post-header-child",
|
|
external=True,
|
|
)
|
|
hdr += (
|
|
f'<div id="root-header-child" class="w-full">'
|
|
f'{post_row}'
|
|
f'</div>'
|
|
)
|
|
|
|
# Error content
|
|
error_html = (
|
|
'<div class="text-center p-8 max-w-lg mx-auto">'
|
|
f'<div class="font-bold text-2xl md:text-4xl text-red-500 mb-4">{errnum}</div>'
|
|
f'<div class="text-stone-600 mb-4">{escape(message)}</div>'
|
|
)
|
|
if image:
|
|
error_html += f'<div class="flex justify-center"><img src="{image}" width="300" height="300"></div>'
|
|
error_html += "</div>"
|
|
|
|
return full_page(ctx, header_rows_html=hdr, content_html=error_html)
|
|
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 = _sexp_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:
|
|
try:
|
|
html = _sexp_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("HX-Request") == "true":
|
|
from shared.sexp.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("HX-Request") == "true":
|
|
from shared.sexp.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("HX-Request") == "true":
|
|
return await make_response(
|
|
"Something went wrong. Please try again.",
|
|
status,
|
|
)
|
|
|
|
# Raw HTML — avoids context processor / fragment loop on errors.
|
|
return await make_response(_error_page(
|
|
"WELL THIS IS EMBARRASSING…"
|
|
), status)
|