Source code for dmr.plugins.msgspec.serializer

from collections.abc import Iterable
from typing import TYPE_CHECKING, Any, ClassVar, Literal

import msgspec
from django.http import HttpRequest
from typing_extensions import TypedDict, 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

if TYPE_CHECKING:
    from dmr.metadata import EndpointMetadata




class ToModelKwargs(TypedDict, total=False):
    """Custom serializer API options, taken by :func:`msgspec.convert`."""

    # `from_attributes` is explicitly left out. It is always `False`.
    builtin_types: Iterable[type] | None
    str_keys: bool





class ToJsonKwargs(ToModelKwargs, total=False):
    """Custom deserializer API options, taken by :func:`msgspec.to_builtins`."""

    order: Literal['deterministic', 'sorted'] | None





class MsgspecEndpointOptimizer(BaseEndpointOptimizer):
    """Optimize endpoints that are parsed with Msgspec."""



    @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.




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] = {}



    @override
    @classmethod
    def serialize(
        cls,
        structure: Any,
        *,
        renderer: Renderer,
    ) -> bytes:
        """Convert any object to a raw bytestring."""
        return renderer.render(structure, cls.serialize_hook)




    @override
    @classmethod
    def deserialize(
        cls,
        buffer: Raw,
        *,
        parser: Parser,
        request: HttpRequest,
        model: Any,
    ) -> Any:
        """Convert string or bytestring to simple python object."""
        return parser.parse(
            buffer,
            cls.deserialize_hook,
            request=request,
            model=model,
        )




    @override
    @classmethod
    def from_python(
        cls,
        unstructured: Any,
        model: Any,
        *,
        strict: bool | 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.

        Returns:
            Structured and validated data.

        Raises:
            msgspec.ValidationError: When parsing can't be done.

        """
        return msgspec.convert(
            unstructured,
            model,
            strict=strict or False,
            dec_hook=cls.deserialize_hook,
            **cls.to_model_kwargs,
        )




    @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,
        )




    @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',
        )