149 lines
5.0 KiB
Python
149 lines
5.0 KiB
Python
import logging
|
|
import traceback
|
|
|
|
from django.conf import settings
|
|
from django.core.exceptions import PermissionDenied, ValidationError as DjangoValidationError
|
|
from django.http import Http404
|
|
from rest_framework import exceptions, status
|
|
from rest_framework.exceptions import APIException
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import exception_handler as drf_exception_handler
|
|
|
|
logger = logging.getLogger("django.request")
|
|
|
|
|
|
def _safe_error_response(
|
|
detail: str,
|
|
code: str,
|
|
status_code: int,
|
|
) -> Response:
|
|
"""Return a consistent error response with no stack traces."""
|
|
return Response(
|
|
{"error": detail, "code": code},
|
|
status=status_code,
|
|
)
|
|
|
|
|
|
def api_exception_handler(exc: Exception, context: dict) -> Response | None:
|
|
"""
|
|
Custom DRF exception handler that:
|
|
- Never exposes stack traces, file paths, or Python internals
|
|
- Maps common exceptions to user-safe messages
|
|
- Logs full traceback to django.request logger
|
|
- Returns consistent {error, code} format
|
|
"""
|
|
# Always log the full traceback
|
|
logger.error(
|
|
"API Exception: %s: %s\n%s",
|
|
type(exc).__name__,
|
|
exc,
|
|
"".join(traceback.format_tb(exc.__traceback__)),
|
|
)
|
|
|
|
# PermissionDenied -> 403
|
|
if isinstance(exc, PermissionDenied):
|
|
return _safe_error_response(
|
|
"You do not have permission to perform this action.",
|
|
"permission_denied",
|
|
status.HTTP_403_FORBIDDEN,
|
|
)
|
|
|
|
# Http404 -> 404
|
|
if isinstance(exc, Http404):
|
|
return _safe_error_response(
|
|
"The requested resource was not found.",
|
|
"not_found",
|
|
status.HTTP_404_NOT_FOUND,
|
|
)
|
|
|
|
# Django ValidationError -> 400
|
|
if isinstance(exc, DjangoValidationError):
|
|
return _safe_error_response(
|
|
str(exc) if isinstance(exc.message, str) else "Validation error.",
|
|
"validation_error",
|
|
status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
|
|
# DRF APIException (includes AuthenticationFailed, NotAuthenticated, ParseError, etc.)
|
|
if isinstance(exc, APIException):
|
|
# Use DRF's standard handling but map to our format
|
|
response = drf_exception_handler(exc, context)
|
|
if response is not None:
|
|
# Ensure response is our safe format
|
|
detail = _extract_detail(response.data)
|
|
return _safe_error_response(
|
|
detail,
|
|
_get_error_code(exc),
|
|
response.status_code,
|
|
)
|
|
|
|
# DRF Throttled
|
|
if isinstance(exc, exceptions.Throttled):
|
|
return _safe_error_response(
|
|
"Request rate limit exceeded. Please try again later.",
|
|
"throttled",
|
|
exc.status_code,
|
|
)
|
|
|
|
# AuthenticationFailed / NotAuthenticated
|
|
if isinstance(exc, exceptions.AuthenticationFailed):
|
|
return _safe_error_response(
|
|
str(exc.detail) if hasattr(exc, "detail") else "Authentication failed.",
|
|
"authentication_failed",
|
|
status.HTTP_401_UNAUTHORIZED,
|
|
)
|
|
|
|
if isinstance(exc, exceptions.NotAuthenticated):
|
|
return _safe_error_response(
|
|
"Authentication credentials were not provided.",
|
|
"not_authenticated",
|
|
status.HTTP_401_UNAUTHORIZED,
|
|
)
|
|
|
|
# Catch-all for unhandled exceptions
|
|
if not settings.DEBUG:
|
|
return _safe_error_response(
|
|
"Internal server error.",
|
|
"internal_error",
|
|
status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
)
|
|
|
|
# In DEBUG mode, let DRF's default handler show the traceback
|
|
return drf_exception_handler(exc, context)
|
|
|
|
|
|
def _extract_detail(data: dict | list | str) -> str:
|
|
"""Extract the first meaningful error string from DRF error data."""
|
|
if isinstance(data, str):
|
|
return data
|
|
if isinstance(data, list):
|
|
for item in data:
|
|
result = _extract_detail(item)
|
|
if result:
|
|
return result
|
|
if isinstance(data, dict):
|
|
# Try 'detail' first, then first field error
|
|
if "detail" in data:
|
|
return _extract_detail(data["detail"])
|
|
for _key, value in data.items():
|
|
result = _extract_detail(value)
|
|
if result:
|
|
return result
|
|
return "An error occurred."
|
|
|
|
|
|
def _get_error_code(exc: APIException) -> str:
|
|
"""Map exception class to a stable error code string."""
|
|
mapping: dict[type, str] = {
|
|
exceptions.AuthenticationFailed: "authentication_failed",
|
|
exceptions.NotAuthenticated: "not_authenticated",
|
|
exceptions.PermissionDenied: "permission_denied",
|
|
exceptions.NotFound: "not_found",
|
|
exceptions.MethodNotAllowed: "method_not_allowed",
|
|
exceptions.NotAcceptable: "not_acceptable",
|
|
exceptions.UnsupportedMediaType: "unsupported_media_type",
|
|
exceptions.Throttled: "throttled",
|
|
exceptions.ParseError: "parse_error",
|
|
exceptions.ValidationError: "validation_error",
|
|
}
|
|
return mapping.get(type(exc), "error") |