import inspect
from collections.abc import Awaitable, Callable, Mapping, Sequence, Set
from functools import wraps
from http import HTTPStatus
from typing import TYPE_CHECKING, Any, ClassVar, Never, overload
from django.http import HttpResponse, HttpResponseBase
from django.urls import URLPattern
from typing_extensions import ParamSpec, Protocol, TypeVar, deprecated
from dmr.cookies import CookieSpec, NewCookie
from dmr.errors import AsyncErrorHandler, SyncErrorHandler
from dmr.exceptions import (
DataRenderingError,
InternalServerError,
NotAuthenticatedError,
ResponseSchemaError,
ValidationError,
)
from dmr.headers import HeaderSpec, NewHeader
from dmr.internal.context import SerializerContext as SerializerContext
from dmr.metadata import EndpointMetadata, ResponseModification, ResponseSpec
from dmr.negotiation import RequestNegotiator, ResponseNegotiator
from dmr.openapi.objects import (
Callback,
ExternalDocumentation,
Link,
Operation,
Reference,
Server,
)
from dmr.parsers import Parser
from dmr.renderers import Renderer
from dmr.response import APIError, APIRedirectError
from dmr.security.base import AsyncAuth, SyncAuth
from dmr.serializer import BaseSerializer
from dmr.settings import HttpSpec, Settings, resolve_setting
from dmr.validation import (
EndpointMetadataBuilder,
EndpointMetadataValidator,
ModifyEndpointPayload,
Payload,
ResponseValidator,
ValidateEndpointPayload,
)
if TYPE_CHECKING:
from dmr.controller import Controller
from dmr.openapi.core.context import OpenAPIContext
from dmr.validation.response import ValidatedModification
[docs]
class Endpoint: # noqa: WPS214
"""
Represents the single API endpoint.
Is built during the import time.
In the runtime only does response validate, which can be disabled.
"""
__slots__ = (
'_func',
'_serializer_context',
'is_async',
'metadata',
'request_negotiator',
'response_negotiator',
'response_validator',
)
# Instance API:
_func: Callable[..., Any]
# Class API:
serializer_context_cls: ClassVar[type[SerializerContext]] = (
SerializerContext
)
metadata_builder_cls: ClassVar[type[EndpointMetadataBuilder]] = (
EndpointMetadataBuilder
)
metadata_validator_cls: ClassVar[type[EndpointMetadataValidator]] = (
EndpointMetadataValidator
)
metadata_cls: ClassVar[type[EndpointMetadata]] = EndpointMetadata
response_modification_cls: ClassVar[type[ResponseModification]] = (
ResponseModification
)
request_negotiator_cls: ClassVar[type[RequestNegotiator]] = (
RequestNegotiator
)
response_negotiator_cls: ClassVar[type[ResponseNegotiator]] = (
ResponseNegotiator
)
response_validator_cls: ClassVar[type[ResponseValidator]] = (
ResponseValidator
)
def __init__(
self,
func: Callable[..., Any],
*,
controller_cls: type['Controller[BaseSerializer]'],
) -> None:
"""
Create an entrypoint.
Args:
func: Entrypoint handler. An actual function to be called.
controller_cls: ``Controller`` class that this endpoint belongs to.
.. danger::
Endpoint object must **not** have any mutable instance state,
because its instance is reused for all requests.
"""
type_annotations = controller_cls.annotations_context(func)
self._serializer_context = self.serializer_context_cls(
func,
controller_cls,
type_annotations,
)
# We need to add payloads to functions that don't have it,
# since decorator is optional:
payload: Payload = getattr(func, '__dmr_payload__', None)
# We add metadata in two steps:
# 1. We construct metadata with no responses yet.
# We only do basic validation at this point: structure, types, etc.
# No semantics validation / etc.
# 2. When metadata is ready, we collect all the responses from all
# of the components that support it. Including custom ones.
# Then we enrich metadata with collected responses and use it.
# Done!
metadata = self.metadata_builder_cls(
payload=payload,
controller_cls=controller_cls,
func=func,
metadata_cls=self.metadata_cls,
response_modification_cls=self.response_modification_cls,
component_parsers=self._serializer_context.component_parsers,
type_annotations=type_annotations,
)()
self.metadata_validator_cls(metadata=metadata)(
func,
payload=payload,
controller_cls=controller_cls,
)
func.__metadata__ = metadata # type: ignore[attr-defined]
self.metadata = metadata
self.request_negotiator = self.request_negotiator_cls(
self.metadata,
controller_cls.serializer,
)
self.response_negotiator = self.response_negotiator_cls(
self.metadata,
controller_cls.serializer,
streaming=controller_cls.streaming,
)
# We need a func before any wrappers, but with metadata:
self.response_validator = self.response_validator_cls(
metadata,
controller_cls.serializer,
)
# We can now run endpoint's optimization:
controller_cls.serializer.optimizer.optimize_endpoint(metadata)
# Now we can add wrappers:
if inspect.iscoroutinefunction(func):
self.is_async = True
self._func = self._async_endpoint(func)
else:
self.is_async = False
self._func = self._sync_endpoint(func)
[docs]
def __call__(
self,
controller: 'Controller[BaseSerializer]',
*args: Any,
**kwargs: Any,
) -> HttpResponseBase:
"""Run the endpoint and return the response."""
return self._func( # type: ignore[no-any-return]
controller,
*args,
**kwargs,
)
[docs]
def handle_error(
self,
controller: 'Controller[BaseSerializer]',
exc: Exception,
) -> HttpResponseBase:
"""
Return error response if possible.
Override this method to add custom error handling.
"""
# NOTE: if you change something here,
# also change in `handle_async_error`
if self.metadata.error_handler is not None:
try:
# We validate this, no error possible in runtime:
return self.metadata.error_handler( # type: ignore[return-value]
self,
controller,
exc,
)
except Exception: # noqa: S110
# We don't use `suppress` here for speed.
pass # noqa: WPS420
# Per-endpoint error handler didn't work.
# Now, try the per-controller one.
try:
return controller.handle_error(
self,
controller,
exc,
)
except Exception:
# And the last option is to handle error globally:
return self._global_error_handler(controller, exc)
[docs]
async def handle_async_error(
self,
controller: 'Controller[BaseSerializer]',
exc: Exception,
) -> HttpResponse:
"""
Return error response if possible.
Override this method to add custom async error handling.
"""
# NOTE: if you change something here, also change in `handle_error`
if self.metadata.error_handler is not None:
try:
# We validate this, no error possible in runtime:
return await self.metadata.error_handler( # type: ignore[no-any-return, misc]
self,
controller,
exc,
)
except Exception: # noqa: S110
# We don't use `suppress` here for speed.
pass # noqa: WPS420
# Per-endpoint error handler didn't work.
# Now, try the per-controller one.
try:
return await controller.handle_async_error(
self,
controller,
exc,
)
except Exception:
# And the last option is to handle error globally:
return self._global_error_handler(controller, exc)
[docs]
def get_schema(
self,
path: str,
pattern: URLPattern,
controller_name: str,
serializer: type[BaseSerializer],
context: 'OpenAPIContext',
) -> Operation:
"""Build an OpenAPI Operation from an endpoint."""
operation_id = self.get_operation_id(
path,
controller_name,
serializer,
context,
)
request_body, params_list = context.generators.component_parsers(
operation_id,
pattern,
self.metadata,
serializer,
)
security = context.generators.security_scheme(
self.metadata.auth,
serializer,
)
return Operation(
tags=self.metadata.tags,
summary=self.metadata.summary,
description=self.metadata.description,
deprecated=self.metadata.deprecated,
security=security,
external_docs=self.metadata.external_docs,
servers=self.metadata.servers,
callbacks=self.metadata.callbacks,
operation_id=operation_id,
request_body=request_body,
responses=context.generators.response(self.metadata, serializer),
parameters=params_list,
)
[docs]
def get_operation_id(
self,
path: str,
controller_name: str,
serializer: type[BaseSerializer],
context: 'OpenAPIContext',
) -> str:
"""Customize how OperationId is generated for the OpenAPI."""
return context.generators.operation_id(
path,
controller_name,
self.metadata,
serializer,
)
def _async_endpoint(
self,
func: Callable[..., Any],
) -> Callable[..., Awaitable[HttpResponseBase]]:
# NOTE: if you change something here, also change in `_sync_endpoint`
@wraps(func)
async def decorator(
controller: 'Controller[BaseSerializer]',
*args: Any,
**kwargs: Any,
) -> HttpResponseBase:
try: # noqa: WPS229
# Negotiate response:
self.response_negotiator(controller.request)
# Run checks:
await self._run_async_checks(controller)
# Parse request:
context = self._serializer_context(self, controller)
# Return response:
func_result = await func(controller, **context)
except (APIError, APIRedirectError) as exc:
func_result = controller.to_error(
exc.raw_data,
status_code=exc.status_code,
headers=exc.headers,
cookies=getattr(exc, 'cookies', None),
)
except Exception as exc:
func_result = await self.handle_async_error(controller, exc)
return self._make_http_response(controller, func_result)
return decorator
def _sync_endpoint(
self,
func: Callable[..., Any],
) -> Callable[..., HttpResponseBase]:
# NOTE: if you change something here, also change in `_async_endpoint`
@wraps(func)
def decorator(
controller: 'Controller[BaseSerializer]',
*args: Any,
**kwargs: Any,
) -> HttpResponseBase:
try: # noqa: WPS229
# Negotiate response:
self.response_negotiator(controller.request)
# Run checks:
self._run_checks(controller)
# Parse request:
context = self._serializer_context(self, controller)
# Return response:
func_result = func(controller, **context)
except (APIError, APIRedirectError) as exc:
func_result = controller.to_error(
exc.raw_data,
status_code=exc.status_code,
headers=exc.headers,
cookies=getattr(exc, 'cookies', None),
)
except Exception as exc:
func_result = self.handle_error(controller, exc)
return self._make_http_response(controller, func_result)
return decorator
def _run_checks(self, controller: 'Controller[BaseSerializer]') -> None:
# Run sync auth checks:
if self.metadata.auth is None:
return
for auth in self.metadata.auth:
assert isinstance(auth, SyncAuth) # noqa: S101
if auth(self, controller) is not None:
return
raise NotAuthenticatedError
async def _run_async_checks(
self,
controller: 'Controller[BaseSerializer]',
) -> None:
# Run async auth checks:
if self.metadata.auth is None:
return
for auth in self.metadata.auth:
assert isinstance(auth, AsyncAuth) # noqa: S101
if (await auth(self, controller)) is not None: # noqa: WPS476
return
raise NotAuthenticatedError
def _make_http_response(
self,
controller: 'Controller[BaseSerializer]',
raw_data: Any | HttpResponse,
) -> HttpResponseBase:
"""
Returns the actual ``HttpResponse`` object after optional validation.
If it is already the :class:`django.http.HttpResponse` object,
just validates it before returning.
"""
try:
return self._validate_response(controller, raw_data)
except ( # noqa: WPS239
ResponseSchemaError,
ValidationError,
DataRenderingError,
InternalServerError,
) as exc:
# We can't call `self.handle_error` or `self.handle_async_error`
# in exception handlers here,
# because it is too late. Since `ResponseSchemaError`
# happened most likely because the return
# schema validation was not successful.
return controller.to_error(
controller.format_error(exc),
status_code=exc.status_code,
)
def _validate_response(
self,
controller: 'Controller[BaseSerializer]',
response_data: Any | HttpResponseBase,
) -> HttpResponseBase:
if isinstance(response_data, HttpResponseBase):
return self.response_validator.validate_response(
self,
controller,
response_data,
)
validated = self.response_validator.validate_modification(
self,
controller,
response_data,
)
return self._build_new_response(controller, validated)
def _build_new_response(
self,
controller: 'Controller[BaseSerializer]',
validated: 'ValidatedModification',
) -> HttpResponseBase:
return controller.to_response(
validated.raw_data,
status_code=validated.status_code,
headers=validated.headers,
cookies=validated.cookies,
renderer=validated.renderer,
)
def _global_error_handler(
self,
controller: 'Controller[BaseSerializer]',
exc: Exception,
) -> HttpResponse:
"""
Import the global error handling and call it.
If not class level error handling has happened.
"""
return resolve_setting( # type: ignore[no-any-return]
Settings.global_error_handler,
import_string=True,
)(self, controller, exc)
_ParamT = ParamSpec('_ParamT')
_ReturnT = TypeVar('_ReturnT')
_ResponseT = TypeVar(
'_ResponseT',
bound=HttpResponseBase | Awaitable[HttpResponseBase],
)
@overload
def validate( # noqa: WPS234
response: ResponseSpec,
/,
*responses: ResponseSpec,
error_handler: AsyncErrorHandler,
validate_responses: bool | None = None,
semantic_responses: bool | None = None,
exclude_semantic_responses: Set[HTTPStatus] | None = frozenset(),
validate_events: bool | None = None,
no_validate_http_spec: Set[HttpSpec] | None = frozenset(),
parsers: Sequence[Parser] | None = None,
renderers: Sequence[Renderer] | None = None,
auth: Sequence[AsyncAuth] | Sequence[SyncAuth] | None = (),
summary: str | None = None,
description: str | None = None,
tags: list[str] | None = None,
operation_id: str | None = None,
deprecated: bool = False,
external_docs: ExternalDocumentation | None = None,
callbacks: dict[str, Callback | Reference] | None = None,
servers: list[Server] | None = None,
) -> Callable[
[Callable[_ParamT, Awaitable[HttpResponseBase]]],
Callable[_ParamT, Awaitable[HttpResponseBase]],
]: ...
@overload
def validate(
response: ResponseSpec,
/,
*responses: ResponseSpec,
error_handler: SyncErrorHandler,
validate_responses: bool | None = None,
semantic_responses: bool | None = None,
exclude_semantic_responses: Set[HTTPStatus] | None = frozenset(),
validate_events: bool | None = None,
no_validate_http_spec: Set[HttpSpec] | None = frozenset(),
parsers: Sequence[Parser] | None = None,
renderers: Sequence[Renderer] | None = None,
auth: Sequence[AsyncAuth] | Sequence[SyncAuth] | None = (),
summary: str | None = None,
description: str | None = None,
tags: list[str] | None = None,
operation_id: str | None = None,
deprecated: bool = False,
external_docs: ExternalDocumentation | None = None,
callbacks: dict[str, Callback | Reference] | None = None,
servers: list[Server] | None = None,
) -> Callable[
[Callable[_ParamT, HttpResponseBase]],
Callable[_ParamT, HttpResponseBase],
]: ...
@overload
def validate(
response: ResponseSpec,
/,
*responses: ResponseSpec,
validate_responses: bool | None = None,
semantic_responses: bool | None = None,
exclude_semantic_responses: Set[HTTPStatus] | None = frozenset(),
validate_events: bool | None = None,
no_validate_http_spec: Set[HttpSpec] | None = frozenset(),
error_handler: None = None,
parsers: Sequence[Parser] | None = None,
renderers: Sequence[Renderer] | None = None,
auth: Sequence[AsyncAuth] | Sequence[SyncAuth] | None = (),
summary: str | None = None,
description: str | None = None,
tags: list[str] | None = None,
operation_id: str | None = None,
deprecated: bool = False,
external_docs: ExternalDocumentation | None = None,
callbacks: dict[str, Callback | Reference] | None = None,
servers: list[Server] | None = None,
) -> Callable[
[Callable[_ParamT, _ResponseT]],
Callable[_ParamT, _ResponseT],
]: ...
[docs]
def validate( # noqa: WPS211 # pyright: ignore[reportInconsistentOverload]
response: ResponseSpec,
/,
*responses: ResponseSpec,
validate_responses: bool | None = None,
semantic_responses: bool | None = None,
exclude_semantic_responses: Set[HTTPStatus] | None = frozenset(),
validate_events: bool | None = None,
no_validate_http_spec: Set[HttpSpec] | None = frozenset(),
error_handler: SyncErrorHandler | AsyncErrorHandler | None = None,
parsers: Sequence[Parser] | None = None,
renderers: Sequence[Renderer] | None = None,
auth: Sequence[AsyncAuth] | Sequence[SyncAuth] | None = (),
summary: str | None = None,
description: str | None = None,
tags: list[str] | None = None,
operation_id: str | None = None,
deprecated: bool = False,
external_docs: ExternalDocumentation | None = None,
callbacks: dict[str, Callback | Reference] | None = None,
servers: list[Server] | None = None,
) -> (
Callable[
[Callable[_ParamT, Awaitable[HttpResponseBase]]],
Callable[_ParamT, Awaitable[HttpResponseBase]],
]
| Callable[
[Callable[_ParamT, HttpResponseBase]],
Callable[_ParamT, HttpResponseBase],
]
):
"""
Decorator to validate responses from endpoints that return ``HttpResponse``.
Apply it to validate important API parts:
.. code:: python
>>> from http import HTTPStatus
>>> from django.http import HttpResponse
>>> from dmr import Controller, validate, ResponseSpec
>>> from dmr.plugins.pydantic import PydanticSerializer
>>> class TaskController(Controller[PydanticSerializer]):
... @validate(
... ResponseSpec(
... return_type=list[int],
... status_code=HTTPStatus.OK,
... ),
... )
... def post(self) -> HttpResponse:
... return HttpResponse(b'[1, 2]', status=HTTPStatus.OK)
Response validation can be disabled for extra speed
by sending *validate_responses* falsy parameter
or by setting this configuration in your ``settings.py`` file:
.. code-block:: python
:caption: settings.py
>>> DMR_SETTINGS = {'validate_responses': False}
Args:
response: The main response that this endpoint is allowed to return.
responses: A collection of other responses that are allowed
to be returned from this endpoint.
validate_responses: Do we have to run runtime validation
of responses for this endpoint? Customizable via global setting,
per controller, and per endpoint.
Here we only store the per endpoint information.
semantic_responses: Should semantic responses be collected
from different providers for this endpoint.
exclude_semantic_responses: Set of semantic responses status codes
that user wants to disable.
validate_events: Should this endpoint validate events?
If not set, defaults to the ``validate_responses`` value.
This value only matters if the response
will be a streaming response that supports event validation.
no_validate_http_spec: Set of http spec validation checks
that we disable for this endpoint.
error_handler: Callback function to be called
when this endpoint faces an exception.
parsers: Sequence of types to be used for this endpoint
to parse incoming request's body. All types must be subtypes
of :class:`~dmr.parsers.Parser`.
renderers: Sequence of types to be used for this endpoint
to render response's body. All types must be subtypes
of :class:`~dmr.renderers.Renderer`.
auth: Sequence of auth instances to be used for this endpoint.
Sync endpoints must use instances
of :class:`dmr.security.SyncAuth`.
Async endpoints must use instances
of :class:`dmr.security.AsyncAuth`.
Set it to ``None`` to disable auth for this endpoint.
summary: A short summary of what the operation does.
description: A verbose explanation of the operation behavior.
tags: A list of tags for API documentation control.
Used to group operations in OpenAPI documentation.
operation_id: Unique string used to identify the operation.
deprecated: Declares this operation to be deprecated.
external_docs: Additional external documentation for this operation.
callbacks: A map of possible out-of band callbacks related to the
parent operation. The key is a unique identifier for the Callback
Object. Each value in the map is a Callback Object that describes
a request that may be initiated by the API provider and the
expected responses.
servers: An alternative servers array to service this operation.
Returns:
The same function with ``__dmr_payload__`` payload instance.
.. warning::
Do not disable ``validate_responses`` unless
this is performance critical for you!
"""
return _add_payload(
payload=ValidateEndpointPayload(
responses=[response, *responses],
validate_responses=validate_responses,
semantic_responses=semantic_responses,
exclude_semantic_responses=exclude_semantic_responses,
validate_events=validate_events,
no_validate_http_spec=no_validate_http_spec,
error_handler=error_handler,
parsers=parsers,
renderers=renderers,
auth=auth,
summary=summary,
description=description,
tags=tags,
operation_id=operation_id,
deprecated=deprecated,
external_docs=external_docs,
callbacks=callbacks,
servers=servers,
),
)
class _ModifyAsyncCallable(Protocol):
"""Make `@modify` on functions returning `HttpResponse` unrepresentable."""
@overload
@deprecated(
# It is not actually deprecated, but impossible for the day one.
# But, this is the only way to trigger a typing error.
'Do not use `@modify` decorator with `HttpResponse` return type',
)
def __call__(self, func: Callable[_ParamT, _ResponseT], /) -> Never: ...
@overload
def __call__(
self,
func: Callable[_ParamT, Awaitable[_ReturnT]],
/,
) -> Callable[_ParamT, _ReturnT]: ...
class _ModifySyncCallable(Protocol):
"""Make `@modify` on functions returning `HttpResponse` unrepresentable."""
@overload
@deprecated(
# It is not actually deprecated, but impossible for the day one.
# But, this is the only way to trigger a typing error.
'Do not use `@modify` decorator with `HttpResponse` return type',
)
def __call__(self, func: Callable[_ParamT, _ResponseT], /) -> Never: ...
@overload
@deprecated(
# It is not actually deprecated, but impossible for the day one.
# But, this is the only way to trigger a typing error.
'Passing sync `error_handler` to `@modify` requires sync endpoint',
)
def __call__(
self,
func: Callable[_ParamT, Awaitable[_ReturnT]],
/,
) -> Never: ...
@overload
def __call__(
self,
func: Callable[_ParamT, _ReturnT],
/,
) -> Callable[_ParamT, _ReturnT]: ...
class _ModifyAnyCallable(Protocol):
"""Make `@modify` on functions returning `HttpResponse` unrepresentable."""
@overload
@deprecated(
# It is not actually deprecated, but impossible for the day one.
# But, this is the only way to trigger a typing error.
'Do not use `@modify` decorator with `HttpResponse` return type',
)
def __call__(self, func: Callable[_ParamT, _ResponseT], /) -> Never: ...
@overload
def __call__(
self,
func: Callable[_ParamT, _ReturnT],
/,
) -> Callable[_ParamT, _ReturnT]: ...
@overload
def modify(
*,
# TODO: make error handlers generic?
error_handler: AsyncErrorHandler,
status_code: HTTPStatus | None = None,
headers: Mapping[str, NewHeader | HeaderSpec] | None = None,
cookies: Mapping[str, NewCookie | CookieSpec] | None = None,
validate_responses: bool | None = None,
semantic_responses: bool | None = None,
exclude_semantic_responses: Set[HTTPStatus] | None = frozenset(),
validate_events: bool | None = None,
extra_responses: list[ResponseSpec] | None = None,
no_validate_http_spec: Set[HttpSpec] | None = frozenset(),
parsers: Sequence[Parser] | None = None,
renderers: Sequence[Renderer] | None = None,
auth: Sequence[AsyncAuth] | Sequence[SyncAuth] | None = (),
summary: str | None = None,
description: str | None = None,
tags: list[str] | None = None,
operation_id: str | None = None,
deprecated: bool = False,
external_docs: ExternalDocumentation | None = None,
callbacks: dict[str, Callback | Reference] | None = None,
servers: list[Server] | None = None,
response_description: str | None = None,
) -> _ModifyAsyncCallable: ...
@overload
def modify(
*,
error_handler: SyncErrorHandler,
status_code: HTTPStatus | None = None,
headers: Mapping[str, NewHeader | HeaderSpec] | None = None,
cookies: Mapping[str, NewCookie | CookieSpec] | None = None,
validate_responses: bool | None = None,
semantic_responses: bool | None = None,
exclude_semantic_responses: Set[HTTPStatus] | None = frozenset(),
validate_events: bool | None = None,
extra_responses: list[ResponseSpec] | None = None,
no_validate_http_spec: Set[HttpSpec] | None = frozenset(),
parsers: Sequence[Parser] | None = None,
renderers: Sequence[Renderer] | None = None,
auth: Sequence[AsyncAuth] | Sequence[SyncAuth] | None = (),
summary: str | None = None,
description: str | None = None,
tags: list[str] | None = None,
operation_id: str | None = None,
deprecated: bool = False,
external_docs: ExternalDocumentation | None = None,
callbacks: dict[str, Callback | Reference] | None = None,
servers: list[Server] | None = None,
links: dict[str, Link | Reference] | None = None,
response_description: str | None = None,
) -> _ModifySyncCallable: ...
@overload
def modify(
*,
status_code: HTTPStatus | None = None,
headers: Mapping[str, NewHeader | HeaderSpec] | None = None,
cookies: Mapping[str, NewCookie | CookieSpec] | None = None,
validate_responses: bool | None = None,
semantic_responses: bool | None = None,
exclude_semantic_responses: Set[HTTPStatus] | None = frozenset(),
validate_events: bool | None = None,
extra_responses: list[ResponseSpec] | None = None,
no_validate_http_spec: Set[HttpSpec] | None = frozenset(),
error_handler: None = None,
parsers: Sequence[Parser] | None = None,
renderers: Sequence[Renderer] | None = None,
auth: Sequence[AsyncAuth] | Sequence[SyncAuth] | None = (),
summary: str | None = None,
description: str | None = None,
tags: list[str] | None = None,
operation_id: str | None = None,
deprecated: bool = False,
external_docs: ExternalDocumentation | None = None,
callbacks: dict[str, Callback | Reference] | None = None,
servers: list[Server] | None = None,
links: dict[str, Link | Reference] | None = None,
response_description: str | None = None,
) -> _ModifyAnyCallable: ...
[docs]
def modify( # noqa: WPS211
*,
status_code: HTTPStatus | None = None,
headers: Mapping[str, NewHeader | HeaderSpec] | None = None,
cookies: Mapping[str, NewCookie | CookieSpec] | None = None,
validate_responses: bool | None = None,
semantic_responses: bool | None = None,
exclude_semantic_responses: Set[HTTPStatus] | None = frozenset(),
validate_events: bool | None = None,
extra_responses: list[ResponseSpec] | None = None,
no_validate_http_spec: Set[HttpSpec] | None = frozenset(),
error_handler: SyncErrorHandler | AsyncErrorHandler | None = None,
parsers: Sequence[Parser] | None = None,
renderers: Sequence[Renderer] | None = None,
auth: Sequence[AsyncAuth] | Sequence[SyncAuth] | None = (),
summary: str | None = None,
description: str | None = None,
tags: list[str] | None = None,
operation_id: str | None = None,
deprecated: bool = False,
external_docs: ExternalDocumentation | None = None,
callbacks: dict[str, Callback | Reference] | None = None,
servers: list[Server] | None = None,
links: dict[str, Link | Reference] | None = None,
response_description: str | None = None,
) -> _ModifyAsyncCallable | _ModifySyncCallable | _ModifyAnyCallable:
"""
Decorator to modify endpoints that return raw model data.
Apply it to change some API parts:
.. code:: python
>>> from http import HTTPStatus
>>> from dmr import Controller, modify
>>> from dmr.plugins.pydantic import PydanticSerializer
>>> class TaskController(Controller[PydanticSerializer]):
... @modify(status_code=HTTPStatus.ACCEPTED)
... def post(self) -> list[int]:
... return [1, 2] # id of tasks you have started
Args:
status_code: Shows *status_code* in the documentation.
When *status_code* is passed, always use it by default.
When not provided, we use smart inference
based on the HTTP method name for default returned response.
headers: Shows *headers* in the documentation.
When *headers* are passed we will add them for the default response.
cookies: Shows *cookies* in the documentation.
When *cookies* are passed we will add them for the default response.
validate_responses: Do we have to run runtime validation
of responses for this endpoint? Customizable via global setting,
per controller, and per endpoint.
Here we only store the per endpoint information.
semantic_responses: Should semantic responses be collected
from different providers for this endpoint.
exclude_semantic_responses: Set of semantic responses status codes
that user wants to disable.
validate_events: Should this endpoint validate events?
If not set, defaults to the ``validate_responses`` value.
This value only matters if the response
will be a streaming response that supports event validation.
extra_responses: List of extra responses that this endpoint can return.
no_validate_http_spec: Set of http spec validation checks
that we disable for this endpoint.
error_handler: Callback function to be called
when this endpoint faces an exception.
parsers: Sequence of types to be used for this endpoint
to parse incoming request's body. All types must be subtypes
of :class:`~dmr.parsers.Parser`.
renderers: Sequence of types to be used for this endpoint
to render response's body. All types must be subtypes
of :class:`~dmr.renderers.Renderer`.
auth: Sequence of auth instances to be used for this endpoint.
Sync endpoints must use instances
of :class:`dmr.security.SyncAuth`.
Async endpoints must use instances
of :class:`dmr.security.AsyncAuth`.
Set it to ``None`` to disable auth for this endpoint.
summary: A short summary of what the operation does.
description: A verbose explanation of the operation behavior.
tags: A list of tags for API documentation control.
Used to group operations in OpenAPI documentation.
operation_id: Unique string used to identify the operation.
deprecated: Declares this operation to be deprecated.
external_docs: Additional external documentation for this operation.
callbacks: A map of possible out-of band callbacks related to the
parent operation. The key is a unique identifier for the Callback
Object. Each value in the map is a Callback Object that describes
a request that may be initiated by the API provider and the
expected responses.
servers: An alternative servers array to service this operation.
links: Possible links to other OpenAPI operations.
response_description: Description for the generated response object.
Returns:
The same function with ``__dmr_payload__`` payload instance.
.. warning::
Do not disable ``validate_responses`` unless
this is performance critical for you!
"""
return _add_payload( # type: ignore[return-value]
payload=ModifyEndpointPayload(
status_code=status_code,
headers=headers,
cookies=cookies,
responses=extra_responses,
validate_responses=validate_responses,
semantic_responses=semantic_responses,
exclude_semantic_responses=exclude_semantic_responses,
validate_events=validate_events,
no_validate_http_spec=no_validate_http_spec,
error_handler=error_handler,
parsers=parsers,
renderers=renderers,
auth=auth,
summary=summary,
description=description,
tags=tags,
operation_id=operation_id,
deprecated=deprecated,
external_docs=external_docs,
callbacks=callbacks,
servers=servers,
links=links,
response_description=response_description,
),
)
def _add_payload(
*,
payload: ModifyEndpointPayload | ValidateEndpointPayload,
) -> Callable[[Callable[_ParamT, _ReturnT]], Callable[_ParamT, _ReturnT]]:
# Add payload for future use in the Endpoint creation.
def decorator(
func: Callable[_ParamT, _ReturnT],
) -> Callable[_ParamT, _ReturnT]:
func.__dmr_payload__ = payload # type: ignore[attr-defined]
return func
return decorator
django-modern-rest