from typing import Optional
from rest_framework import status
from rest_framework.response import Response
from rest_framework.views import exception_handler
from certego_saas.apps.payments.consts import STRIPE_SENSITIVE_FIELDS_SET
from .log import LogBuilder
__all__ = [
"custom_exception_handler",
]
[docs]def custom_exception_handler(exc, context) -> Optional[Response]:
"""
Extends ``rest_framework.views.exception_handler``.
- Custom handlers for NotFound, ValidationError, AuthenticationError
- Uses :class:`certego_saas.ext.log.LogBuilder` for JSON logging.
"""
# If an exception is thrown that we don't explicitly handle here, we want
# to delegate to the default exception handler offered by DRF. If we do
# handle this exception type, we will still want access to the response
# generated by DRF, so we get that response up front.
response = exception_handler(exc, context)
handlers = {
"NotFound": _handle_not_found_error,
"ValidationError": _handle_generic_error,
"AuthenticationError": _handle_stripe_error,
}
# This is how we identify the type of the current exception. We will use
# this in a moment to see whether we should handle this exception or let
# Django REST Framework do it's thing.
exc_class = exc.__class__.__name__
exc_class_base = exc.__class__.__base__.__name__
if exc_class in handlers:
handler = handlers[exc_class]
response = handler(exc, context, response)
elif exc_class_base in handlers:
handler = handlers[exc_class_base]
response = handler(exc, context, response)
if _should_log(exc, context):
try:
log_builder = LogBuilder(
request=context.get("request", None),
response=response,
view=context.get("view", None),
exception=exc,
) # log exception in JSON format
log_builder.sensitive_fields = STRIPE_SENSITIVE_FIELDS_SET
log_builder.build_log()
log_builder.handle_log()
except Exception:
pass
else:
# see django.utils.log.log_response
response._has_been_logged = True # type: ignore
return response
def _handle_generic_error(exc, context, response) -> Response:
# This is about the most straightforward exception handler we can create.
# We take the response generated by DRF and wrap it in the `errors` key.
if hasattr(response, "data"):
response.data = {"errors": response.data}
return response
def _handle_not_found_error(exc, context, response) -> Response:
view = context.get("view", None)
if view and hasattr(view, "queryset") and view.queryset is not None:
error_key = view.queryset.model._meta.verbose_name
response.data = {"errors": {error_key: response.data["detail"]}}
else:
response = _handle_generic_error(exc, context, response)
return response
def _handle_stripe_error(exc, context, response) -> Response:
return Response(
{
"errors": f"Failed to fetch {context['view'].get_view_name()}. Please try again later."
},
status.HTTP_500_INTERNAL_SERVER_ERROR,
)
def _should_log(exc, context) -> bool:
"""
Reference: https://github.com/certego/Dragonfly/issues/856
"""
flag = True
view = context.get("view", None)
if view:
code = getattr(exc, "status_code", None)
viewname = view.__class__.__name__
if (code == 404 or exc.__class__.__name__ == "Http404") and viewname in [
"APIAccessTokenView",
"OrganizationViewSet",
]:
flag = False
return flag