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

src/app/errorhandler.py
78 строк3 KB
Artem Ivochkin
todo со ссылками на модуль с кодом
11 янв 2026, 18:45
11 янв 2026, 18:453ad959e
# -*- coding: utf-8 -*-
"""ВНИМАНИЕ
   Если FastAPI.debug == True, то перехват Exception подавляется и управление из FastAPI сюда не попадает
"""
# todo выяснить, почему не получается сделать универсальный обработчик, внутри которого определять исключения
import logging

from fastapi import Request
from fastapi.responses import PlainTextResponse

from serv.log_utils import format_flatten_dict, trunc_str

logger = logging.getLogger(__name__)


async def request_validation_error(req: Request, exc):
    """Валидация запроса.

       FastApi вызывает Pydantic
       - съедает исключение Pydantic-а ValidationError,
       - добавляет свои обработки и
       - перевозбуждает исключение под именем RequestValidationError
    """
    response = PlainTextResponse(status_code=400, content='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 = PlainTextResponse(status_code=576, content='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 = PlainTextResponse(status_code=stat, content=resp)

    return response