src/app/errorhandler.py - master - Ulyanoff/mlops2

src/app/errorhandler.py
93 строки4 KB
Artem Ivochkin
тесты электронного болвана (user add & del)
11 дек 2025, 11:05
11 дек 2025, 11:05745c950
# -*- coding: utf-8 -*-
import logging

from fastapi import Request
from starlette.responses import JSONResponse

from serv.log_utils import trunc_str

logger = logging.getLogger(__name__)

"""ВНИМАНИЕ 
   Если FastAPI.debug == True, то перехват Exception подавляется и управление из FastAPI сюда не попадает
"""


async def request_validation_error(req: Request, exc):
    """Валидация запроса"""
    """FastApi вызывает Pydantic
       - съедает исключение Pydantic-а ValidationError,
       - добавляет свои обработки и
       - перевозбуждает исключение под именем RequestValidationError
    """
    response = JSONResponse(status_code=400, content={'error': 'Bad request, wrong parameter(s)'})

    ll, ls = f'\n\t<~\t\tRequest validation error: {req.url.path}, {req.scope["endpoint"].__name__}\n', list()
    for error in exc.errors():
        ls.append(f"\t\t\t- {str(error.get('type'))[:16]:<16} "
                  f"{str(error.get('loc')):<20} in {str(error.get('input')):<12} {error.get('msg')}")
        if error.get('type', None) == 'json_invalid': ls[-1] += f"\n\t\t\t  {error.get('ctx', None)}"

    ll += '\n'.join(ls)
    logger.error(ll)

    return response


async def response_validation_error(req: Request, exc):
    """Валидация ответа"""

    response = JSONResponse(status_code=576, content={'error': 'Wrong server data'})

    ll, ls = f'\n\t<~\t\tResponse validation error: {req.url.path}, {req.scope["endpoint"].__name__}\n', list()
    for error in exc.errors():
        ls.append(f"\t\t\t- {str(error.get('type'))[:16]:<16} "
                  f"{str(error.get('loc')):<20} in {str(error.get('input')):<12} {error.get('msg')}")
        if error.get('type', None) == 'json_invalid': ls[-1] += f"\n\t\t\t  {error.get('ctx', None)}"

    ll += '\n'.join(ls)
    logger.error(ll)

    return response


async def sql_programming_error(req: Request, exc):
    """Обработка ошибок базы"""
    stat, resp = 576, "Wrong server data"

    ll = (f'\n\t<~\t\tSQLAlchemy {getattr(type(exc), "__name__", "Error")} handler: {req.url.path},'
          f' {req.scope["endpoint"].__name__}')
    if getattr(exc, 'code', None) == 'f405':
        if getattr(exc, 'orig', None):
            if getattr(exc.orig, 'sqlstate') == '42P01':
                resp = "Table not found"
                if len(exc.orig.args) > 0:
                    err_msg = exc.orig.args[0].split(': ')[-1].capitalize()
                    ll += f'\n\t\t\tUnsuccessful attempt database reading.' \
                          f'\n\t\t\t{err_msg}, ' \
                          f'{getattr(exc.__context__, "__dict__") if hasattr(exc, "__context__") else "Unknown"}'
    else:
        ll += trunc_str(f"{exc}")

    ll += (f'\n\t<=\t\t{req.url.path}, {req.scope["endpoint"].__name__} HTTP={stat} failed')
    logger.error(ll)

    response = JSONResponse(status_code=stat, content={"error": resp})

    return response


async def db_integrity_error(req: Request, exc):
    """Обработка ошибок целостности базы"""
    ll = f'\n\t<~\t\tDatabase integrity error: {req.url.path}, {req.scope["endpoint"].__name__}\n'

    if "email" in str(exc.orig).lower():
        stat, msg = 409, "Email already registered"
    else:
        stat, msg = 422, "Database integrity error"

    ll += f"\t\t\t{trunc_str(exc)}"
    ll += (f'\n\t<=\t\t{req.url.path}, {req.scope["endpoint"].__name__} HTTP={stat} failed')
    logger.error(ll)

    return JSONResponse(status_code=stat, content={"error": msg})