from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, ClassVar, Literal
import msgspec
from django.http import HttpRequest
from typing_extensions import TypedDict, TypeForm, override
from dmr.errors import ErrorDetail, ErrorType
from dmr.parsers import Parser, Raw
from dmr.plugins.msgspec.schema import MsgspecSchemaGenerator
from dmr.renderers import Renderer
from dmr.serializer import (
BaseEndpointOptimizer,
BaseSerializer,
ContextField,
ContextModel,
context_field_tuples,
)
if TYPE_CHECKING:
from dmr.metadata import EndpointMetadata
class _CommonKwargs(TypedDict, total=False):
# `from_attributes` is explicitly left out. It is always `False`.
builtin_types: Iterable[type[Any]] | None
str_keys: bool
[docs]
class ToModelKwargs(_CommonKwargs, total=False, closed=True):
"""Custom serializer API options, taken by :func:`msgspec.convert`."""
[docs]
class ToJsonKwargs(_CommonKwargs, total=False, closed=True):
"""Custom deserializer API options, taken by :func:`msgspec.to_builtins`."""
order: Literal['deterministic', 'sorted'] | None
[docs]
class MsgspecEndpointOptimizer(BaseEndpointOptimizer):
"""Optimize endpoints that are parsed with Msgspec."""
__slots__ = ()
[docs]
@override
@classmethod
def optimize_endpoint(cls, metadata: 'EndpointMetadata') -> None:
"""Does nothing for msgspec."""
# `msgspec.convert` does not have any API
# to pre-build validation schema.
# Returning `Struct` or `list[Struct]` will be just fast enough.
# Creating first `Encoder` and `Decoder` objects are fast enough.
[docs]
class MsgspecSerializer(BaseSerializer):
"""
Serialize and deserialize objects using msgspec.
Msgspec support is optional.
To install it run:
.. code:: bash
pip install 'django-modern-rest[msgspec]'
Attributes:
to_json_kwargs: Dictionary of kwargs that will be passed
to model serialization callbacks.
to_model_kwargs: Dictionary of kwargs that will be passed
to model deserialization callbacks.
"""
__slots__ = ()
# Required API:
validation_error = msgspec.ValidationError
optimizer = MsgspecEndpointOptimizer
schema_generator = MsgspecSchemaGenerator
# Custom API:
to_json_kwargs: ClassVar[ToJsonKwargs] = {}
to_model_kwargs: ClassVar[ToModelKwargs] = {}
[docs]
@override
@classmethod
def serialize(
cls,
structure: Any,
*,
renderer: Renderer,
) -> bytes:
"""Convert any object to a raw bytestring."""
return renderer.render(structure, cls.serialize_hook)
[docs]
@override
@classmethod
def deserialize(
cls,
buffer: Raw,
*,
parser: Parser,
request: HttpRequest,
model: TypeForm[Any],
) -> Any:
"""Convert string or bytestring to simple python object."""
return parser.parse(
buffer,
cls.deserialize_hook,
request=request,
model=model,
)
[docs]
@override
@classmethod
def from_python(
cls,
unstructured: Any,
model: TypeForm[Any],
*,
strict: bool | None,
extra_namespace: Mapping[str, Any] | None = None,
) -> Any:
"""
Parse *unstructured* data from python primitives into *model*.
Args:
unstructured: Python objects to be parsed / validated.
model: Python type to serve as a model.
Can be any type that ``msgspec`` supports.
Examples: ``dict[str, int]`` and ``BaseModel`` subtypes.
strict: Whether we use more strict validation rules.
For example, it is fine for a request validation
to be less strict in some cases and allow type coercition.
But, response types need to be strongly validated.
extra_namespace: Not used currently.
Returns:
Structured and validated data.
Raises:
msgspec.ValidationError: When parsing can't be done.
.. versionchanged:: 0.13.0
Added *extra_namespace* parameter.
"""
return msgspec.convert(
unstructured,
model,
strict=strict or False,
dec_hook=cls.deserialize_hook,
**cls.to_model_kwargs,
)
[docs]
@override
@classmethod
def build_context_model(
cls,
name: str,
fields: Mapping[str, ContextField],
) -> ContextModel:
"""
Build the model to parse the whole request context at once.
We always build a :class:`msgspec.Struct` with ``gc=False``,
it is around x2 faster to validate than a :class:`typing.TypedDict`.
Struct instances are converted into keyword arguments
with :func:`msgspec.structs.asdict`.
Defaults are passed as-is, so ``msgspec`` rules apply:
mutable defaults like ``[1]`` or non-frozen structs are not allowed.
.. versionadded:: 0.16.0
"""
model = msgspec.defstruct(
name,
context_field_tuples(fields),
# Instances live only for a single request and never have
# any cycles, there's no need to track them:
gc=False,
)
return ContextModel(model, to_kwargs=msgspec.structs.asdict)
[docs]
@override
@classmethod
def to_python(
cls,
structured: Any,
) -> Any:
"""
Unparse *structured* data from a model into Python primitives.
Args:
structured: Model instance.
Returns:
Unstructured data.
"""
return msgspec.to_builtins(
structured,
enc_hook=cls.serialize_hook,
**cls.to_json_kwargs,
)
[docs]
@override
@classmethod
def serialize_validation_error(
cls,
exc: Exception,
) -> list[ErrorDetail]:
"""Serialize validation error."""
if isinstance(exc, msgspec.ValidationError):
return [{'msg': str(exc), 'type': str(ErrorType.value_error)}]
raise NotImplementedError(
f'Cannot serialize exception {exc!r} of type {type(exc)} safely',
)
django-modern-rest