96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
import logging
|
|
import traceback
|
|
from collections.abc import Iterable
|
|
from typing import Any
|
|
|
|
from django.conf import settings
|
|
from rest_framework import status as http_status
|
|
from rest_framework.exceptions import ErrorDetail
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import exception_handler as drf_exception_handler
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _flatten_messages(values: Iterable) -> list[str]:
|
|
items: list[str] = []
|
|
for value in values:
|
|
items.extend(_to_str_list(value))
|
|
return items
|
|
|
|
|
|
def _to_str_list(value: str | ErrorDetail | list | tuple | dict) -> list[str]:
|
|
if isinstance(value, str | ErrorDetail):
|
|
return [str(value)]
|
|
if isinstance(value, list | tuple):
|
|
return _flatten_messages(value)
|
|
if isinstance(value, dict):
|
|
items: list[str] = []
|
|
for field, v in value.items():
|
|
msgs = _to_str_list(v)
|
|
for msg in msgs:
|
|
if field in ("non_field_errors", "__all__"):
|
|
items.append(str(msg))
|
|
else:
|
|
items.append(f"{field}: {msg}")
|
|
return items
|
|
return [str(value)]
|
|
|
|
|
|
def _format_payload(messages: list[str], status_code: int) -> dict[str, Any]:
|
|
clean_messages: list[str] = []
|
|
for msg in messages:
|
|
msg = msg.replace("error:", "").strip()
|
|
if ":" in msg:
|
|
_, only_msg = msg.split(":", 1)
|
|
clean_messages.append(only_msg.strip())
|
|
else:
|
|
clean_messages.append(msg)
|
|
|
|
error_message = messages[0] if messages else http_status.HTTP_STATUS_CODES.get(status_code, "Error")
|
|
|
|
return {
|
|
"error": error_message,
|
|
"status_code": status_code,
|
|
"messages": [{"message": msg} for msg in clean_messages],
|
|
}
|
|
|
|
|
|
def _request_extra(context: dict[str, Any]) -> dict[str, Any]:
|
|
request = context.get("request")
|
|
meta = getattr(request, "META", {})
|
|
return {
|
|
"request_method": getattr(request, "method", None),
|
|
"request_url": getattr(request, "get_full_path", lambda: None)(),
|
|
"remote_addr": meta.get("REMOTE_ADDR") if meta else None,
|
|
"user_agent": meta.get("HTTP_USER_AGENT", "") if meta else "",
|
|
}
|
|
|
|
|
|
def exception_handler(exc, context) -> Response:
|
|
response = drf_exception_handler(exc, context)
|
|
is_server_error = response is None or getattr(response, "status_code", 500) >= 500
|
|
if is_server_error:
|
|
logger.exception("DRF exception", extra=_request_extra(context))
|
|
if settings.DEBUG:
|
|
is_unhandled = response is None
|
|
if is_unhandled or is_server_error:
|
|
raise
|
|
|
|
if response is not None:
|
|
status_code = response.status_code
|
|
detail = response.data
|
|
if status_code < 500:
|
|
messages = _to_str_list(detail)
|
|
payload = _format_payload(messages, status_code)
|
|
return Response(payload, status=status_code)
|
|
|
|
traceback_text = traceback.format_exc()
|
|
payload = _format_payload(
|
|
["Internal server error."],
|
|
http_status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
)
|
|
payload["exception"] = str(exc)
|
|
payload["traceback"] = traceback_text
|
|
return Response(payload, status=http_status.HTTP_500_INTERNAL_SERVER_ERROR)
|