import enum
import inspect
from collections.abc import Awaitable, Callable
from functools import wraps
from typing import (
TYPE_CHECKING,
Any,
Final,
NotRequired,
TypeAlias,
final,
overload,
)
from django.http import HttpResponse
from django.utils.encoding import force_str
from typing_extensions import TypedDict
from dmr.exceptions import (
DataRenderingError,
InternalServerError,
NotAcceptableError,
NotAuthenticatedError,
RequestSerializationError,
ResponseSchemaError,
ValidationError,
)
if TYPE_CHECKING:
from dmr.controller import Controller
from dmr.endpoint import Endpoint
from dmr.serializer import BaseSerializer
[docs]
@final
@enum.unique
class ErrorType(enum.StrEnum):
"""
Collection of all possible error types that we use in DMR.
Attributes:
value_error: Raised when we can't parse something.
internal_error: Raised when internal error happens.
not_allowed: Raised when using unsupported http method. 405 alias.
security: Raised when security related error happens.
user_msg: Raised for custom errors from users.
not_found: Raised when we can't find controller.
streaming: Happens when we stream events.
"""
value_error = 'value_error'
internal_error = 'internal_error'
not_allowed = 'not_allowed'
security = 'security'
user_msg = 'user_msg'
not_found = 'not_found'
streaming = 'streaming'
[docs]
class ErrorDetail(TypedDict):
"""Base schema for error details description."""
msg: str
type: NotRequired[str]
loc: NotRequired[list[int | str]]
[docs]
class ErrorModel(TypedDict):
"""
Default error response schema.
Can be customized.
See :ref:`customizing-error-messages` for more details.
"""
detail: list[ErrorDetail]
#: Error handler type for sync callbacks.
SyncErrorHandler: TypeAlias = Callable[
['Endpoint', 'Controller[BaseSerializer]', Exception], # noqa: WPS226
HttpResponse,
]
#: Error handler type for async callbacks.
AsyncErrorHandler: TypeAlias = Callable[
['Endpoint', 'Controller[BaseSerializer]', Exception],
Awaitable[HttpResponse],
]
_MethodSyncHandler: TypeAlias = Callable[
# This is not `Any`, this a `Controller[BaseSerializer]` instance,
# but mypy can't do better:
['Any', 'Endpoint', 'Controller[Any]', Exception],
HttpResponse,
]
_MethodAsyncHandler: TypeAlias = Callable[
# This is not `Any`, this a `Controller[BaseSerializer]` instance,
# but mypy can't do better:
['Any', 'Endpoint', 'Controller[Any]', Exception],
Awaitable[HttpResponse],
]
@overload
def wrap_handler(method: _MethodSyncHandler) -> SyncErrorHandler: ...
@overload
def wrap_handler(method: _MethodAsyncHandler) -> AsyncErrorHandler: ...
[docs]
def wrap_handler(
method: _MethodSyncHandler | _MethodAsyncHandler,
) -> SyncErrorHandler | AsyncErrorHandler:
"""
Utility function to wrap controller methods.
It is used to wrap an existing controller method
and pass it as ``error_handler=`` argument to an endpoint.
"""
if inspect.iscoroutinefunction(method):
@wraps(method)
async def decorator( # pyright: ignore[reportRedeclaration]
endpoint: 'Endpoint',
controller: 'Controller[BaseSerializer]',
exc: Exception,
) -> HttpResponse:
return await method( # type: ignore[no-any-return]
controller,
endpoint,
controller,
exc,
)
else:
@wraps(method) # pyrefly: ignore[bad-argument-type]
def decorator(
endpoint: 'Endpoint',
controller: 'Controller[BaseSerializer]',
exc: Exception,
) -> HttpResponse:
return method( # type: ignore[return-value]
controller,
endpoint,
controller,
exc,
)
return decorator
# NOTE: keep this tuple in sync with `format_error()`
_default_handled_excs: Final = (
RequestSerializationError,
ResponseSchemaError, # can only happen if validation is enabled
NotAuthenticatedError,
NotAcceptableError,
ValidationError,
InternalServerError,
DataRenderingError,
)
[docs]
def global_error_handler(
endpoint: 'Endpoint',
controller: 'Controller[BaseSerializer]',
exc: Exception,
) -> HttpResponse:
"""
Global error handler for all cases.
It is the last item in the chain that we try:
1. Per endpoint configuration via
:meth:`~dmr.endpoint.Endpoint.handle_error`
and :meth:`~dmr.endpoint.Endpoint.handle_async_error`
methods
2. Per controller handlers
3. This global handler, specified via the configuration
If some exception cannot be handled, it is just reraised.
Args:
endpoint: Endpoint where error happened.
controller: Controller instance that *endpoint* belongs to.
exc: Exception instance that happened.
Returns:
:class:`~django.http.HttpResponse` with proper response for this error.
Or raise *exc* back.
Here's an example that will produce
``{'detail': [{'msg': 'inf', 'type': 'user_msg'}]}``
for any :exc:`ZeroDivisionError` in your application:
.. code:: python
>>> from http import HTTPStatus
>>> from django.http import HttpResponse
>>> from dmr.controller import Controller
>>> from dmr.endpoint import Endpoint
>>> from dmr.errors import global_error_handler, ErrorType
>>> def custom_error_handler(
... controller: Controller,
... endpoint: Endpoint,
... exc: Exception,
... ) -> HttpResponse:
... if isinstance(exc, ZeroDivisionError):
... return controller.to_error(
... controller.format_error(
... 'inf',
... error_type=ErrorType.user_msg,
... ),
... status_code=HTTPStatus.NOT_IMPLEMENTED,
... )
... # Call the original handler to handle default errors:
... return global_error_handler(controller, endpoint, exc)
>>> # And then in your settings file:
>>> DMR_SETTINGS = {
... # Object `custom_error_handler` will also work:
... 'global_error_handler': 'path.to.custom_error_handler',
... }
.. warning::
Make sure you always call original ``global_error_handler``
in the very end. Unless, you want to disable original error handling.
"""
if isinstance(exc, _default_handled_excs):
return controller.to_error(
controller.format_error(exc),
status_code=exc.status_code,
)
raise exc from None
django-modern-rest