devplacepy/devplacepy/errors.py

# retoor <retoor@molodetz.nl>

import logging

from fastapi import Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse, RedirectResponse

from devplacepy import happy404
from devplacepy.responses import json_error, wants_json
from devplacepy.schemas import ValidationErrorOut
from devplacepy.seo import base_seo_context
from devplacepy.services.moderation.screening import ContentRefused
from devplacepy.templating import templates
from devplacepy.utils import safe_next

logger = logging.getLogger(__name__)


async def not_found(request: Request, exc):
    if wants_json(request):
        return json_error(404, "Not found")
    happy_response = happy404.render(request)
    if happy_response is not None:
        return happy_response
    seo_ctx = base_seo_context(
        request,
        title="Not Found - DevPlace",
        description="The page you requested does not exist.",
        robots="noindex",
    )
    return templates.TemplateResponse(
        request,
        "error.html",
        {
            **seo_ctx,
            "request": request,
            "error_code": 404,
            "error_message": "Page not found",
        },
        status_code=404,
    )


async def server_error(request: Request, exc):
    logger.exception("500 error on %s %s", request.method, request.url.path)
    if wants_json(request):
        return json_error(500, "Internal server error")
    seo_ctx = base_seo_context(
        request,
        title="Server Error - DevPlace",
        description="Something went wrong.",
        robots="noindex",
    )
    return templates.TemplateResponse(
        request,
        "error.html",
        {
            **seo_ctx,
            "request": request,
            "error_code": 500,
            "error_message": "Internal server error",
        },
        status_code=500,
    )


async def content_refused(request: Request, exc: ContentRefused):
    logger.info("content refused on %s %s: %s", request.method, request.url.path, exc.message)
    if wants_json(request):
        return json_error(400, exc.message, categories=list(exc.categories))
    seo_ctx = base_seo_context(
        request,
        title="Content not published - DevPlace",
        description=exc.message,
        robots="noindex",
    )
    return templates.TemplateResponse(
        request,
        "error.html",
        {
            **seo_ctx,
            "request": request,
            "error_code": 400,
            "error_message": exc.message,
        },
        status_code=400,
    )


_AUTH_FORM_PAGES = {
    "/auth/signup": ("signup.html", "Join DevPlace"),
    "/auth/login": ("login.html", "Sign In"),
    "/auth/forgot-password": ("forgot_password.html", "Reset Password"),
}

_FRIENDLY_ERRORS = {
    ("username", "too_short"): "Username must be between 3 and 32 characters",
    ("username", "too_long"): "Username must be between 3 and 32 characters",
    ("password", "too_short"): "Password must be at least 6 characters",
}


def _friendly_error(err):
    field = err["loc"][-1] if err.get("loc") else ""
    key = (field, err.get("type", "").replace("string_", ""))
    if key in _FRIENDLY_ERRORS:
        return _FRIENDLY_ERRORS[key]
    msg = err.get("msg", "Invalid input")
    prefix = "Value error, "
    return msg[len(prefix) :] if msg.startswith(prefix) else msg


async def on_validation_error(request: Request, exc: RequestValidationError):
    errors = [_friendly_error(e) for e in exc.errors()]
    if wants_json(request):
        fields: dict = {}
        for raw, friendly in zip(exc.errors(), errors):
            name = raw["loc"][-1] if raw.get("loc") else "_"
            fields.setdefault(str(name), []).append(friendly)
        return JSONResponse(
            ValidationErrorOut(fields=fields, messages=errors).model_dump(mode="json"),
            status_code=422,
        )
    path = request.url.path
    page = _AUTH_FORM_PAGES.get(path)
    if page is None and path.startswith("/auth/reset-password/"):
        page = ("reset_password.html", "Set New Password")
    if page:
        template_name, title = page
        context = {
            **base_seo_context(request, title=title, robots="noindex,nofollow"),
            "request": request,
            "errors": errors,
        }
        try:
            form = await request.form()
            context.update({k: v for k, v in form.items() if isinstance(v, str)})
        except Exception:
            pass
        if "token" in request.path_params:
            context["token"] = request.path_params["token"]
        return templates.TemplateResponse(
            request, template_name, context, status_code=400
        )
    referer = safe_next(request.headers.get("referer"), "/feed")
    return RedirectResponse(url=referer, status_code=303)


def register(app):
    app.exception_handler(404)(not_found)
    app.exception_handler(500)(server_error)
    app.exception_handler(ContentRefused)(content_refused)
    app.exception_handler(RequestValidationError)(on_validation_error)