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 (
""
""
"Error"
""
""
)
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:
# 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.sexp.helpers import (
root_header_sexp, post_header_sexp,
header_child_sexp, full_page_sexp, sexp_call,
)
hdr = root_header_sexp(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 = post_header_sexp(ctx)
if post_row:
hdr = "(<> " + hdr + " " + header_child_sexp(post_row) + ")"
# Error content
error_content = sexp_call("error-content", errnum=errnum, message=message, image=image)
return full_page_sexp(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 = _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("SX-Request") == "true" or 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("SX-Request") == "true" or 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 {escape(service)} 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,
)
# Raw HTML — avoids context processor / fragment loop on errors.
return await make_response(_error_page(
"WELL THIS IS EMBARRASSING…"
), status)
"
""
"