from collections.abc import Callable, Mapping, Sequence, Set
from http import HTTPMethod, HTTPStatus
from typing import TYPE_CHECKING, Any, ClassVar, Final, Generic, TypeVar
from django.http import HttpRequest, HttpResponse, HttpResponseBase
from django.utils.functional import classproperty
from django.utils.translation import gettext_lazy as _
from django.views import View
from typing_extensions import TypeForm, override
from typing_extensions import deprecated as typing_deprecated
from dmr import throttling as dmr_throttling
from dmr.cookies import NewCookie
from dmr.endpoint import Endpoint, Extras
from dmr.errors import ErrorModel, ErrorType, format_error
from dmr.exceptions import EndpointMetadataError, UnsolvableAnnotationsError
from dmr.internal.docstrings import resolve_summary_and_description
from dmr.internal.io import identity
from dmr.internal.types import EMPTY, StrOrPromise
from dmr.metadata import ResponseSpec
from dmr.negotiation import request_renderer
from dmr.openapi.collector import InternalRouteMetadata
from dmr.openapi.core.context import OpenAPIContext
from dmr.openapi.objects import (
Callback,
ExternalDocumentation,
Operation,
PathItem,
Reference,
SecurityRequirement,
Server,
)
from dmr.parsers import Parser
from dmr.renderers import Renderer
from dmr.response import build_response
from dmr.security.base import AsyncAuth, SyncAuth
from dmr.serializer import BaseSerializer
from dmr.settings import HttpSpec
from dmr.types import AnnotationsContext, infer_type_args
from dmr.validation import ControllerValidator, SettingsValidator
if TYPE_CHECKING:
from dmr.routing import Router
_METHOD_NOT_ALLOWED_MSG: Final = _(
'Method {method} is not allowed, allowed: {allowed}',
)
_SerializerT_co = TypeVar(
'_SerializerT_co',
bound=BaseSerializer,
covariant=True,
)
_ResponseT = TypeVar('_ResponseT', bound=HttpResponse)
[docs]
class Controller(View, Generic[_SerializerT_co]): # noqa: WPS214
"""
Defines API views as controllers.
Controller is a :class:`django.views.generic.base.View` subclass
that should be used as a base for all REST endpoints.
Attributes:
endpoint_cls: Class to create endpoints with.
serializer: Serializer that is passed via type parameters.
The main goal of the serializer is to serialize object
to json and deserialize them from json.
You can't change the serializer simply by modifying
the attribute in the controller class.
Because it is already passed to many other places.
To customize it: create a new class,
subclass :class:`~dmr.serializer.BaseSerializer`,
and pass the new type as a type argument to the controller.
settings_validator_cls: Runs settings validation
once the first controller is created.
no_validate_http_spec: Set of http spec validation checks
that we disable for this class.
Overrides the settings value, can be overridden per endpoint.
Set it to ``None`` to enable all checks back.
validate_responses: Boolean whether or not validating responses.
Works in runtime, can be disabled for better performance.
exclude_validate_responses: Set of status codes that we don't
validate, even when ``validate_responses`` is enabled.
Useful for errors like ``500`` that can be raised
from anywhere and that you might not want to describe.
Overrides the settings value, can be overridden per endpoint.
Set it to ``None`` to validate all status codes back.
semantic_schema: Should we generate any semantic schema
for endpoints in this controller?
semantic_responses: Should semantic responses be collected
from different providers for all endpoints in this class.
exclude_semantic_responses: Set of semantic responses
that user wants to disable.
Overrides the settings value, can be overridden per endpoint.
Set it to ``None`` to enable all semantic responses back.
semantic_auth: Should semantic auth be collected
from different providers for this endpoint.
Overrides settings value.
exclude_semantic_auth: Set of semantic security requirements names
that should not be collected. Overrides settings value.
responses: List of responses schemas that this controller can return.
Overrides ``'responses'`` key in the settings,
can be overridden per endpoint.
Set it to ``None`` to not use any responses from the settings.
allowed_http_methods: Set of names to be treated as names for endpoints.
Does not include ``options``, but includes ``meta``.
parsers: Sequence of parsers to be used for this controller
to parse incoming request's body. All instances must be of subtypes
of :class:`~dmr.parsers.Parser`.
Overrides the settings value, can be overridden per endpoint.
renderers: Sequence of renderers to be used for this controller
to render response's body. All instances must be of subtypes
of :class:`~dmr.renderers.Renderer`.
Overrides the settings value, can be overridden per endpoint.
validate_negotiation: Should we validate that returned response's
``Content-Type`` header matches the one
that we inferred in the negotiation process?
auth: Sequence of auth instances to be used for this controller.
Sync controllers must use instances
of :class:`dmr.security.SyncAuth`.
Async controllers must use instances
of :class:`dmr.security.AsyncAuth`.
Overrides the settings value, can be overridden per endpoint.
Set it to ``None`` to disable auth of this controller.
throttling: Sequence of throttle instances to be used.
Sync controllers must use instances
of :class:`dmr.throttling.SyncThrottle`.
Async controllers must use instances
of :class:`dmr.throttling.AsyncThrottle`.
Overrides the settings value, can be overridden per endpoint.
Set it to ``None`` to disable throttling of this controller.
error_model: Schema type that represents
and validates common error responses.
is_abstract: Whether or not this controller is abstract.
We consider controller "abstract" when it does not have
exact serializer type or exact ``api_endpoints`` instances.
You can also set it to ``True`` explicitly to make
a controller with an exact serializer reusable
without routing it: it does not build any endpoints,
only its subclasses do. Subclasses that don't declare
``is_abstract`` themselves become concrete again.
Abstract controllers cannot be routed, ``as_view`` raises
:class:`~dmr.exceptions.EndpointMetadataError` for them.
is_async: Whether or not this controller is async.
streaming: Does this controller work with streaming responses like SSE?
extras: Default extras instance for this controller.
Setting it enables ``extras=`` in ``@modify`` and ``@validate``
for all endpoints and provides controller-level defaults.
``EMPTY`` means that extras are not supported.
See :ref:`modify-and-validate-with-extras` to learn more.
controller_validator_cls: Runs full controller validation on definition.
annotations_context: Inference context to call
:func:`typing.get_type_hints` for this controller.
api_endpoints: Dictionary of HTTPMethod name to controller instance.
csrf_exempt: Should this controller be exempted from the CSRF check?
Is ``True`` by default. See :ref:`controller-csrf`
to configure the CSRF correctly to support REST responses.
It is only supported on the controller level, because
it has its own per-method logic
inside Django's original CSRF middleware.
login_required: Whether this controller should be handled by
Django's ``LoginRequiredMiddleware``.
Is ``False`` by default.
Users should make use of authentication in ``django-modern-rest``.
See :doc:`/pages/auth/common` for more details.
summary: A short summary of what this path item does.
Defaults to the first paragraph of the controller's docstring.
Set it to ``None`` to have no summary at all.
description: A verbose explanation of the path item behavior.
Defaults to everything that goes after the first paragraph
of the controller's docstring.
Set it to ``None`` to have no description at all.
tags: A list of tags to group all operations
from this controller in OpenAPI documentation.
Overrides router-level tags, can be overridden per endpoint.
Set it to ``None`` to have no tags at all.
security: A sequence of security requirement objects for all operations
in this controller. Overrides ``security``
of :class:`dmr.openapi.OpenAPIConfig`, can be overridden
per endpoint. Set it to ``None``
to disable ``security`` for this controller.
It is merged with the requirements generated from ``auth``,
but it never affects the runtime auth.
Useful to document external security mechanisms,
for example, the ones enforced by an HTTP proxy.
Used security schemes must be declared
as ``security_schemes`` in ``components``
of :class:`dmr.openapi.OpenAPIConfig`.
See :ref:`customizing_security_openapi`.
servers: An alternative servers array to service all operations
in this controller. Can be overridden per endpoint.
Set it to ``None`` to have no servers at all.
deprecated: Declares all operations in this controller
to be deprecated. Overrides router-level value,
can be overridden per endpoint.
external_docs: Additional external documentation
for all operations in this controller.
Can be overridden per endpoint.
Set it to ``None`` to have no external docs at all.
callbacks: A map of possible out-of band callbacks related
to all operations in this controller.
Can be overridden per endpoint.
Set it to ``None`` to have no callbacks at all.
x_extensions: Specification extensions
of the :class:`~dmr.openapi.objects.PathItem`
of this controller, keys must start with ``x-``.
Unlike other OpenAPI attributes, it is not a default
for the operations: every OpenAPI object has its own
extensions, so nothing is inherited or merged.
Use ``x_extensions`` of :func:`~dmr.endpoint.modify`
or :func:`~dmr.endpoint.validate` for an operation.
ignore_from_spec: If set to ``True``, all endpoints from this controller
would not be added to the final OpenAPI spec.
request: Current :class:`~django.http.HttpRequest` instance.
args: Path positional parameters of the request.
kwargs: Path named parameters of the request.
.. versionchanged:: 0.16.0
Explicit ``is_abstract`` definitions are now respected:
abstract controllers do not build any endpoints.
Added ``deprecated``, ``external_docs``, ``callbacks``,
and ``x_extensions``.
``servers`` is now resolved per endpoint
and dumped on operations, not on the path item.
"""
# Public class-level API:
controller_validator_cls: ClassVar[type[ControllerValidator]] = (
ControllerValidator
)
settings_validator_cls: ClassVar[type[SettingsValidator]] = (
SettingsValidator
)
api_endpoints: ClassVar[Mapping[str, Endpoint]]
csrf_exempt: ClassVar[bool] = True
login_required: ClassVar[bool] = False
serializer: ClassVar[type[BaseSerializer]]
endpoint_cls: ClassVar[type[Endpoint]] = Endpoint
no_validate_http_spec: ClassVar[Set[HttpSpec] | EMPTY | None] = EMPTY
validate_responses: ClassVar[bool | EMPTY] = EMPTY
exclude_validate_responses: ClassVar[Set[HTTPStatus] | EMPTY | None] = EMPTY
semantic_schema: ClassVar[bool | EMPTY] = EMPTY
semantic_responses: ClassVar[bool | EMPTY] = EMPTY
exclude_semantic_responses: ClassVar[Set[HTTPStatus] | EMPTY | None] = EMPTY
semantic_auth: bool | EMPTY = EMPTY
exclude_semantic_auth: Set[str] | EMPTY | None = EMPTY
responses: ClassVar[Sequence[ResponseSpec] | EMPTY | None] = EMPTY
allowed_http_methods: ClassVar[Set[str]] = frozenset(
# We replace old existing `View.options` method with modern `meta`:
{method.name.lower() for method in HTTPMethod} - {'options'} | {'meta'},
)
parsers: ClassVar[Sequence[Parser] | EMPTY] = EMPTY
renderers: ClassVar[Sequence[Renderer] | EMPTY] = EMPTY
validate_negotiation: ClassVar[bool | EMPTY] = EMPTY
auth: ClassVar[Sequence[SyncAuth] | Sequence[AsyncAuth] | EMPTY | None] = (
EMPTY
)
throttling: ClassVar[
Sequence[dmr_throttling.SyncThrottle]
| Sequence[dmr_throttling.AsyncThrottle]
| EMPTY
| None
] = EMPTY
error_model: ClassVar[TypeForm[Any]] = ErrorModel
is_abstract: ClassVar[bool] = True
is_async: ClassVar[bool | None] = None # `None` means that nothing's found
streaming: ClassVar[bool] = False
extras: ClassVar[Extras[Any] | EMPTY] = EMPTY
annotations_context: ClassVar[AnnotationsContext] = AnnotationsContext()
# OpenAPI:
summary: ClassVar[StrOrPromise | EMPTY | None] = EMPTY
description: ClassVar[StrOrPromise | EMPTY | None] = EMPTY
tags: ClassVar[Sequence[str] | EMPTY | None] = EMPTY
servers: ClassVar[Sequence[Server] | EMPTY | None] = EMPTY
security: ClassVar[Sequence[SecurityRequirement] | EMPTY | None] = EMPTY
deprecated: ClassVar[bool | EMPTY] = EMPTY
external_docs: ClassVar[ExternalDocumentation | EMPTY | None] = EMPTY
callbacks: ClassVar[Mapping[str, Callback | Reference] | EMPTY | None] = (
EMPTY
)
x_extensions: ClassVar[Mapping[str, Any] | None] = None
ignore_from_spec: ClassVar[bool] = False
# Public instance API:
kwargs: dict[str, Any]
[docs]
@override
def __init_subclass__(cls) -> None:
"""Construct a controller."""
super().__init_subclass__()
serializer = cls._infer_serializer()
if serializer is None:
return # this is an abstract controller
cls.serializer = serializer
cls.settings_validator_cls(serializer=cls.serializer)()
# Explicit `is_abstract` definitions always win. We don't use
# `getattr`, because subclasses of explicitly abstract controllers
# must become concrete again,
# unless they declare `is_abstract` themselves.
explicit_is_abstract = cls.__dict__.get('is_abstract', False)
if explicit_is_abstract:
# Abstract controllers have nothing to serve,
# so endpoints are only built in a concrete context:
cls.api_endpoints = {}
cls.is_abstract = True
else:
# Now it is validated that we don't have intersections.
cls.api_endpoints = {
canonical: cls.endpoint_cls(
meth,
controller_cls=cls,
)
for canonical, meth in cls._find_existing_http_methods().items()
}
# A controller that has no endpoints is abstract either way:
cls.is_abstract = not bool(cls.api_endpoints)
cls.is_async = cls.controller_validator_cls()(cls)
[docs]
@override
@classmethod
def as_view(cls, **initkwargs: Any) -> Callable[..., HttpResponseBase]:
"""
Returns a view function for the class-based view.
This override applies CSRF exemption to the view. Session-based
authentication will still be explicitly validated for CSRF,
while all other authentication methods will be CSRF-exempt.
This override also applies whether a login is required or not.
By default, login is not required
to exempt the view from Django's ``LoginRequiredMiddleware``.
Users should make use of authentication in ``django-modern-rest``.
See :doc:`/pages/auth/common` for more details.
Raises:
EndpointMetadataError: When called on an abstract controller,
because it has nothing to serve.
.. versionchanged:: 0.16.0
Abstract controllers now raise
:class:`~dmr.exceptions.EndpointMetadataError`
instead of silently returning a broken view.
Controllers now set ``login_required`` to ``False`` by default
in order to exempt controllers from
Django's ``LoginRequiredMiddleware``.
"""
if cls.is_abstract:
raise EndpointMetadataError(
f'{cls!r} is abstract, it cannot be used as a view. '
'Controllers are abstract when they do not have '
'an exact serializer type or any endpoints. '
'Use a subclass with a real serializer '
'and at least one endpoint',
)
# We don't use `csrf_exempt()` decorator here, because it is slow:
view = super().as_view(**initkwargs)
if cls.csrf_exempt:
view.csrf_exempt = True # type: ignore[attr-defined]
# Apply login requirement to the view.
# By default, login is not required to exempt the view
# from Django's `LoginRequiredMiddleware`.
view.login_required = cls.login_required # type: ignore[attr-defined]
return view
[docs]
@override
def setup(self, request: HttpRequest, *args: Any, **kwargs: Any) -> None:
"""
Set request context.
Unlike :meth:`~django.views.generic.base.View.setup` does not set
``head`` method automatically.
Thread safety: there's only one controller instance per request.
"""
self.request = request
self.args = args
self.kwargs = kwargs
[docs]
@override
def dispatch(
self,
request: HttpRequest,
*args: Any,
**kwargs: Any,
) -> HttpResponseBase:
"""
Find an endpoint that serves this HTTP method and call it.
Return 405 if this method is not allowed.
"""
# Fast path for method resolution:
method: str = request.method # type: ignore[assignment]
endpoint = self.api_endpoints.get(method)
if endpoint is not None:
return endpoint.func(self, *args, **kwargs)
# This return is very special,
# since it does not have an attached endpoint.
# All other responses are handled on endpoint level
# with all the response type validation.
return self.handle_method_not_allowed(method)
[docs]
def to_response(
self,
raw_data: Any,
*,
status_code: HTTPStatus | None = None,
headers: Mapping[str, str] | None = None,
cookies: Mapping[str, NewCookie] | None = None,
renderer: Renderer | None = None,
) -> HttpResponse:
"""
Helpful method to convert response parts into an actual response.
Should be always used instead of using
raw :class:`django.http.HttpResponse` objects.
Has better serialization speed and semantics than manual.
Does the usual validation, no "second validation" problem exists.
"""
# For mypy: this can't be `None` at this point.
assert self.request.method # noqa: S101
return build_response(
self.serializer,
method=self.request.method,
raw_data=raw_data,
headers=headers,
cookies=cookies,
status_code=status_code,
renderer=renderer or request_renderer(self.request),
)
[docs]
def to_error(
self,
raw_data: Any,
*,
status_code: HTTPStatus,
headers: Mapping[str, str] | None = None,
cookies: Mapping[str, NewCookie] | None = None,
renderer: Renderer | None = None,
) -> HttpResponse:
"""
Helpful method to convert API error parts into an actual error.
Always requires the error code to be passed.
Is an alias for ``to_response`` method with a siglightly different
signature and semantics.
Should be always used instead of using
raw :class:`django.http.HttpResponse` objects.
Does the usual validation, no "second validation" problem exists.
"""
return self.to_response(
raw_data,
headers=headers,
cookies=cookies,
status_code=status_code,
renderer=(
renderer
or request_renderer(
self.request,
use_nonstreaming_renderer=True,
)
),
)
[docs]
def handle_error(
self,
endpoint: Endpoint,
controller: 'Controller[_SerializerT_co]',
exc: Exception,
) -> HttpResponse:
"""
Return error response if possible. Sync case.
Override this method to add custom error handling for sync execution.
By default - does nothing, only re-raises the passed error.
Won't be called when using async endpoints.
"""
raise # noqa: PLE0704
[docs]
async def handle_async_error(
self,
endpoint: Endpoint,
controller: 'Controller[_SerializerT_co]',
exc: Exception,
) -> HttpResponse:
"""
Return error response if possible. Async case.
Override this method to add custom error handling for async execution.
By default - does nothing, only re-raises the passed error.
Won't be called when using sync endpoints.
"""
raise # noqa: PLE0704
[docs]
@override
@typing_deprecated(
# It is not actually deprecated, but type checkers have no other
# ways to raise custom errors.
'Please do not use this method with `django-modern-rest`, '
'use `handle_method_not_allowed` instead',
)
def http_method_not_allowed(
self,
request: HttpRequest,
*args: Any,
**kwargs: Any,
) -> HttpResponse:
"""
Do not use, use :meth:`handle_method_not_allowed` instead.
``View.http_method_not_allowed`` raises an error in a wrong format.
"""
raise NotImplementedError(
'Please do not use this method with `django-modern-rest`, '
'use `handle_method_not_allowed` instead',
)
[docs]
@override
@typing_deprecated(
# It is not actually deprecated, but type checkers have no other
# ways to raise custom errors.
'Please do not use `options` method with `django-modern-rest`, '
'define your own `meta` method instead',
)
def options(
self,
request: HttpRequest,
*args: Any,
**kwargs: Any,
) -> HttpResponse:
"""
Do not use, define your own `meta` method instead.
Django's `View.options` has incompatible signature with
``django-modern-rest``. It would be a typing error
to define something like:
.. warning::
Don't do this!
.. code:: python
>>> from http import HTTPStatus
>>> from dmr import Controller, validate
>>> from dmr.plugins.pydantic import (
... PydanticSerializer,
... )
>>> class MyController(Controller[PydanticSerializer]):
... @validate(
... ResponseSpec(
... None,
... status_code=HTTPStatus.NO_CONTENT,
... ),
... )
... def options(self) -> HttpResponse: # <- typing problem
... ...
That's why instead of ``options`` you should define
our own ``meta`` method:
.. code:: python
>>> class MyController(Controller[PydanticSerializer]):
... @validate(
... ResponseSpec(
... None,
... status_code=HTTPStatus.NO_CONTENT,
... ),
... )
... def meta(self) -> HttpResponse:
... allow = ','.join(
... method.upper()
... for method in self.allowed_http_methods
... )
... return self.to_response(
... None,
... status_code=HTTPStatus.NO_CONTENT,
... headers={'Allow': allow},
... )
.. note::
By default ``meta`` method is not provided for you.
If you want to support ``OPTIONS`` http method
with the default implementation, use:
.. code:: python
>>> from dmr.options_mixins import MetaMixin
>>> class ControllerWithMeta(
... MetaMixin,
... Controller[PydanticSerializer],
... ): ...
"""
raise NotImplementedError(
'Please do not use `options` method with `django-modern-rest`, '
'define your own `meta` method instead',
)
[docs]
def handle_method_not_allowed(
self,
method: str,
) -> HttpResponse:
"""
Return error response for 405 response code.
It is special in way that we don't have an endpoint associated with it.
"""
# This method cannot call `self.to_response`, because it does not have
# an endpoint associated with it. We switch to lower level
# `build_response` primitive
allowed_methods = sorted(self.api_endpoints.keys())
# NOTE: this response is not validated, so be careful with the spec!
return self._maybe_wrap(
build_response(
self.serializer,
raw_data=self.format_error(
_METHOD_NOT_ALLOWED_MSG.format(
method=repr(method),
allowed=repr(allowed_methods),
),
error_type=ErrorType.not_allowed,
),
status_code=HTTPStatus.METHOD_NOT_ALLOWED,
headers={'Allow': ', '.join(allowed_methods)},
renderer=request_renderer(self.request),
),
)
[docs]
@classmethod
def get_schema( # noqa: WPS210
cls,
route_metadata: InternalRouteMetadata,
context: OpenAPIContext,
router: 'Router',
) -> PathItem | None:
"""
Generate OpenAPI spec for path items.
.. versionchanged:: 0.13.0
- Renamed from ``get_path_item`` to ``get_schema``
- Now allows to return ``None`` to ignore whole
path items from OpenAPI schema
.. versionchanged:: 0.16.0
``summary`` and ``description`` are now parsed
from the controller's docstring when they are not set explicitly.
Changed *path* and *pattern* parameters
to be *route_metadata* instead.
``servers`` is not set on the path item anymore,
it is resolved per endpoint instead.
"""
assert not cls.is_abstract, f"Can't include abstract controller: {cls}" # noqa: S101
operations: dict[str, Operation] = {}
for method, endpoint in cls.api_endpoints.items():
if endpoint.metadata.ignore_from_spec:
continue
operations[method.lower()] = endpoint.get_schema(
route_metadata,
cls,
context,
router,
)
if not operations:
return None
standard_ops, additional_ops = PathItem.split_operations(operations)
summary, description = resolve_summary_and_description(
cls.__doc__,
cls.summary,
cls.description,
)
return PathItem(
**standard_ops,
additional_operations=additional_ops,
summary=summary,
description=description,
x_extensions=(
None if cls.x_extensions is None else dict(cls.x_extensions)
),
)
@classproperty
@override
def view_is_async(cls) -> bool: # noqa: N805 # pyright: ignore[reportIncompatibleVariableOverride] # pyrefly: ignore[bad-override]
"""We already know this in advance, no need to recalculate."""
# This is a part of the `django.View` API, so it must be there.
return cls.is_async is True
# Protected API:
@classmethod
def _infer_serializer(cls) -> type[_SerializerT_co] | None:
existing_serializer: type[_SerializerT_co] | None = getattr(
cls,
'serializer',
None,
)
if existing_serializer is not None:
# It was already set by a base class. See #873
return existing_serializer
type_args = infer_type_args(cls, Controller)
if not type_args:
raise UnsolvableAnnotationsError(
f'Type args {type_args} are not correct for {cls}, '
'at least 1 type arg must be provided',
)
serializer = type_args[0]
if isinstance(serializer, TypeVar):
return None # This is a generic subclass of a controller.
if (
not isinstance(serializer, type)
or not issubclass(serializer, BaseSerializer)
or serializer is BaseSerializer
):
raise UnsolvableAnnotationsError(
f'Type arg {serializer} is not correct for {cls}, '
'it must be a BaseSerializer subclass',
)
return serializer # type: ignore[return-value]
@classmethod
def _maybe_wrap(
cls,
response: _ResponseT,
) -> _ResponseT:
"""Wraps response into a coroutine if this is an async controller."""
if cls.is_async:
return identity(response)
return response
@classmethod
def _find_existing_http_methods(cls) -> dict[str, Callable[..., Any]]:
"""
Returns what HTTP methods are implemented in this controller.
Returns both canonical http method name and our dsl name.
"""
return {
# Rename `meta` back to `options`:
('OPTIONS' if dsl_method == 'meta' else dsl_method.upper()): method
# Sorted, because set order depends on `PYTHONHASHSEED`,
# and endpoints must have the same order in every process:
for dsl_method in sorted(cls.allowed_http_methods)
if (method := getattr(cls, dsl_method, None)) is not None
}
django-modern-rest