ef806f8fbb
Phase 1-3 of decoupling plan: - Shared DB, models, infrastructure, browser, config, utils - Event infrastructure (domain_events outbox, bus, processor) - Structured logging - Generic container concept (container_type/container_id) - Alembic migrations for all schema changes Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
127 lines
3.9 KiB
Python
127 lines
3.9 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
|
|
|
|
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 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:
|
|
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 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":
|
|
# Build a little styled <ul><li>...</li></ul> snippet
|
|
lis = "".join(
|
|
f"<li>{escape(m)}</li>"
|
|
for m in messages if m
|
|
)
|
|
html = (
|
|
"<ul class='list-disc pl-5 space-y-1 text-sm text-red-600'>"
|
|
f"{lis}"
|
|
"</ul>"
|
|
)
|
|
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(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":
|
|
# Generic message for unexpected/untrusted errors
|
|
return await make_response(
|
|
"Something went wrong. Please try again.",
|
|
status,
|
|
)
|
|
|
|
html = await render_template("_types/root/exceptions/error.html")
|
|
return await make_response(html, status)
|