import dataclasses
from http import HTTPStatus
from operator import attrgetter
from typing import TYPE_CHECKING, Literal
from dmr.internal.types import EMPTY
from dmr.openapi.mappers.example import generate_example, set_generated_example
from dmr.openapi.objects import (
Header,
MediaType,
Reference,
Response,
Responses,
Schema,
)
if TYPE_CHECKING:
from dmr.controller import Controller
from dmr.headers import HeaderSpec
from dmr.metadata import EndpointMetadata, ResponseSpec
from dmr.openapi.core.context import OpenAPIContext
from dmr.serializer import BaseSerializer
[docs]
@dataclasses.dataclass(frozen=True, slots=True)
class ResponseGenerator:
"""Generator for OpenAPI ``Response`` objects."""
_context: 'OpenAPIContext'
[docs]
def __call__(
self,
metadata: 'EndpointMetadata',
controller_cls: type['Controller[BaseSerializer]'],
) -> Responses:
"""
Generate responses from response specs.
.. versionchanged:: 0.16.0
Now accepts *controller_cls* parameter instead of *serializer*.
"""
return {
# Delegate call to `ResponseSpec`, so it can change
# how the spec is generated.
str(status_code.value): response_spec.get_schema(
metadata,
controller_cls,
self._context,
)
# Sorted by status code, not by the definition order:
for status_code, response_spec in sorted(
metadata.responses.items(),
)
}
[docs]
def get_schema( # noqa: WPS211
self,
response_spec: 'ResponseSpec',
metadata: 'EndpointMetadata',
controller_cls: type['Controller[BaseSerializer]'],
context: 'OpenAPIContext',
*,
schema_field_name: Literal['schema', 'item_schema'] = 'schema',
used_for_response: bool = True,
content_schema: Schema | None = None,
) -> Response:
"""
Returns the OpenAPI schema for the response.
Can be customized in ``ResponseSpec`` subclasses.
When *content_schema* is passed, it is used for all content types
instead of the schema generated from the return type.
.. versionchanged:: 0.16.0
Added *content_schema* parameter.
"""
headers: dict[str, Header | Reference] = {}
headers.update(
self._get_headers(
response_spec,
metadata,
controller_cls,
context,
),
)
headers.update(
self._get_cookies(
response_spec,
metadata,
controller_cls,
context,
),
)
return Response(
description=(
HTTPStatus(response_spec.status_code).phrase
if response_spec.description is None
else str(response_spec.description)
),
summary=(
None
if response_spec.summary is None
else str(response_spec.summary)
),
links=(
None
if response_spec.links is None
else dict(response_spec.links)
),
x_extensions=(
None
if response_spec.x_extensions is None
else dict(response_spec.x_extensions)
),
# Sorted by header name, not by the definition order:
headers=dict(sorted(headers.items())) or None,
content=self._get_content(
response_spec,
metadata,
controller_cls,
context,
schema_field_name=schema_field_name,
used_for_response=used_for_response,
content_schema=content_schema,
),
)
def _get_headers(
self,
response_spec: 'ResponseSpec',
metadata: 'EndpointMetadata',
controller_cls: type['Controller[BaseSerializer]'],
context: 'OpenAPIContext',
) -> dict[str, Header | Reference]:
if not response_spec.headers:
return {}
return {
name: self._get_header(header_spec, controller_cls, context)
for name, header_spec in response_spec.headers.items()
}
def _get_header(
self,
header_spec: 'HeaderSpec',
controller_cls: type['Controller[BaseSerializer]'],
context: 'OpenAPIContext',
) -> Header:
schema = context.generators.schema(str, controller_cls.serializer)
header = Header(
description=(
None
if header_spec.description is None
else str(header_spec.description)
),
deprecated=header_spec.deprecated or None,
required=header_spec.required or None,
schema=schema,
)
if header_spec.example is not None:
# for mypy: `str` cannot return a reference, it is a primitive
assert isinstance(schema, Schema) # noqa: S101
# Examples written by hand replace the generated ones:
header.example = header_spec.example
header.schema = dataclasses.replace(schema, examples=None)
return header
def _get_cookies(
self,
response_spec: 'ResponseSpec',
metadata: 'EndpointMetadata',
controller_cls: type['Controller[BaseSerializer]'],
context: 'OpenAPIContext',
) -> dict[str, Header | Reference]:
if not response_spec.cookies:
return {}
cookies: dict[str, Header | Reference] = {}
for name, cookie_spec in response_spec.cookies.items():
schema = context.generators.schema(str, controller_cls.serializer)
# for mypy: `str` cannot return a reference, it is a primitive
assert isinstance(schema, Schema) # noqa: S101
# A `Set-Cookie` value is `name=value`, so we only generate
# the value part. `replace` copies the schema, so the example
# does not land on the shared `str` one:
cookie_value = generate_example(str, controller_cls.serializer)
schema = set_generated_example(
dataclasses.replace(schema),
EMPTY if cookie_value is EMPTY else f'{name}={cookie_value}',
)
cookies[f'Set-Cookie: {name}'] = Header(
description=(
None
if cookie_spec.description is None
else str(cookie_spec.description)
),
required=cookie_spec.required or None,
schema=schema,
)
return cookies
def _get_content( # noqa: WPS211
self,
response_spec: 'ResponseSpec',
metadata: 'EndpointMetadata',
controller_cls: type['Controller[BaseSerializer]'],
context: 'OpenAPIContext',
*,
schema_field_name: str,
used_for_response: bool,
content_schema: Schema | None,
) -> dict[str, MediaType | Reference]:
# Import cycle:
from dmr.internal.negotiation import ( # noqa: PLC0415
get_conditional_types,
)
return_types = (
get_conditional_types(response_spec.return_type, ()) or {}
)
return {
# Sorted by content type, not by the renderers order:
renderer.content_type: MediaType(
**{ # type: ignore[arg-type]
schema_field_name: (
context.generators.schema(
return_types.get(
renderer.content_type,
response_spec.return_type,
),
controller_cls.serializer,
used_for_response=used_for_response,
)
if content_schema is None
else content_schema
),
},
)
for renderer in sorted(
metadata.renderers.values(),
key=attrgetter('content_type'),
)
if (
not response_spec.limit_to_content_types
or renderer.content_type in response_spec.limit_to_content_types
)
}
django-modern-rest