151 lines
5.3 KiB
Python
151 lines
5.3 KiB
Python
# DRF Rack
|
|
# Библиотека классов расширений для Django REST Framework
|
|
#
|
|
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
|
|
# Все права защищены. Лицензия: MIT
|
|
|
|
"""
|
|
Описание функции custom_exception_handler
|
|
"""
|
|
|
|
import logging
|
|
|
|
from rest_framework import status
|
|
from rest_framework.exceptions import MethodNotAllowed
|
|
from rest_framework.exceptions import NotAuthenticated
|
|
from rest_framework.exceptions import PermissionDenied
|
|
from rest_framework.exceptions import ValidationError
|
|
from rest_framework.request import Request
|
|
from rest_framework.response import Response
|
|
|
|
from net.xeaf.rack.core import CoreException
|
|
from net.xeaf.rack.models.responses import ErrorResponse
|
|
from net.xeaf.rack.utils.exceptions import ForbiddenException
|
|
from net.xeaf.rack.utils.exceptions import NotFoundException
|
|
from net.xeaf.rack.utils.exceptions import UnauthorizedException
|
|
from wsgiapp import settings
|
|
|
|
|
|
def custom_exception_handler(exc: CoreException | Exception, context: dict) -> Response:
|
|
"""
|
|
Универсальный обработчик исключений
|
|
|
|
:param exc: Исключение
|
|
:param context: Контекст
|
|
|
|
:return: Ответ
|
|
"""
|
|
|
|
# Получаем информацию о запросе
|
|
request = context.get("request")
|
|
assert isinstance(request, Request)
|
|
|
|
# Это наше исключение
|
|
if isinstance(exc, CoreException):
|
|
return _make_error_response(
|
|
status_code=exc.status_code,
|
|
detail=exc.detail,
|
|
meta=exc.meta,
|
|
pure_status=exc.is_pure_status(),
|
|
exc_info=exc)
|
|
|
|
# Проверка на стандартное исключение валидации
|
|
if isinstance(exc, ValidationError):
|
|
return _process_validation_exception(exc)
|
|
|
|
# Проверка на стандартное исключение авторизации
|
|
if isinstance(exc, NotAuthenticated):
|
|
return _make_error_response(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=UnauthorizedException.Msg.DEFAULT_MESSAGE,
|
|
meta=None,
|
|
pure_status=False,
|
|
exc_info=exc)
|
|
|
|
# Проверка на стандартное исключение прав доступа
|
|
if isinstance(exc, PermissionDenied):
|
|
return _make_error_response(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=ForbiddenException.Msg.DEFAULT_MESSAGE,
|
|
meta=None,
|
|
pure_status=False,
|
|
exc_info=exc)
|
|
|
|
# Проверка на стандартное исключение отсутствия реализации
|
|
if isinstance(exc, MethodNotAllowed):
|
|
return _make_error_response(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=NotFoundException.Msg.DEFAULT_MESSAGE,
|
|
meta=None,
|
|
pure_status=True,
|
|
exc_info=exc)
|
|
|
|
# Если ничего другое не прошло
|
|
return _make_error_response(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(exc),
|
|
meta=None,
|
|
pure_status=True,
|
|
exc_info=exc)
|
|
|
|
|
|
def _process_validation_exception(exc: ValidationError) -> ErrorResponse:
|
|
"""
|
|
Обрабатывает частный случай стандартного исключения валидации
|
|
|
|
:param exc: Исключение
|
|
|
|
:return: Ответ
|
|
"""
|
|
|
|
# Обработка списка
|
|
if isinstance(exc.detail, list):
|
|
if len(exc.detail) == 1:
|
|
detail = str(exc.detail[0])
|
|
else:
|
|
detail = exc.detail
|
|
|
|
# Обработка словаря
|
|
elif isinstance(exc.detail, dict):
|
|
detail = {}
|
|
for field, error_info in exc.detail.items():
|
|
if isinstance(error_info, list):
|
|
error = error_info[0]
|
|
detail[field] = str(error)
|
|
|
|
# Обработка остальных типов
|
|
else:
|
|
detail = str(exc.detail)
|
|
|
|
return _make_error_response(status_code=status.HTTP_400_BAD_REQUEST, detail=detail, meta=None, pure_status=False, exc_info=exc)
|
|
|
|
|
|
def _make_error_response(status_code: int, detail: list | dict | str | bool | None, meta: dict | None, pure_status: bool,
|
|
exc_info: Exception) -> ErrorResponse:
|
|
"""
|
|
Создает ответ с ошибкой
|
|
|
|
:param status_code: Код статуса
|
|
:param detail: Детали ошибки
|
|
:param meta: Дополнительная информация
|
|
:param pure_status: Признак отправки чистого кода состояния
|
|
:param exc_info: Исключение
|
|
|
|
:return: Ответ
|
|
"""
|
|
|
|
logger = logging.getLogger(settings.LOGGER_NAME)
|
|
|
|
if status_code == status.HTTP_400_BAD_REQUEST:
|
|
logger.debug(str(exc_info), exc_info=exc_info)
|
|
elif status_code == status.HTTP_401_UNAUTHORIZED:
|
|
logger.warning(str(exc_info), exc_info=exc_info)
|
|
elif status_code == status.HTTP_403_FORBIDDEN:
|
|
logger.warning(str(exc_info), exc_info=exc_info)
|
|
elif status_code == status.HTTP_404_NOT_FOUND:
|
|
logger.error(str(exc_info), exc_info=exc_info)
|
|
else:
|
|
logger.critical(str(exc_info), exc_info=exc_info)
|
|
|
|
return ErrorResponse(status_code=status_code, detail=detail, meta=meta, pure_status=pure_status)
|