import dataclasses
from collections.abc import Callable, Mapping
from functools import lru_cache
from typing import (
TYPE_CHECKING,
Any,
ClassVar,
Final,
Literal,
TypeAlias,
TypeVar,
final,
)
import pydantic
import pydantic_core
from django.http import HttpRequest
from pydantic.config import ExtraValues
from typing_extensions import TypedDict, TypeForm, override
from dmr.envs import MAX_CACHE_SIZE
from dmr.errors import ErrorDetail, ErrorType
from dmr.exceptions import (
DataParsingError,
DataRenderingError,
EndpointMetadataError,
)
from dmr.internal.json import NativeJson, json_loads
from dmr.parsers import Parser, Raw
from dmr.plugins.pydantic.schema import PydanticSchemaGenerator
from dmr.renderers import Renderer
from dmr.serializer import (
BaseEndpointOptimizer,
BaseSerializer,
ContextField,
ContextModel,
context_field_tuples,
)
from dmr.types import EMPTY
if TYPE_CHECKING:
from dmr.controller import Controller
from dmr.metadata import EndpointMetadata
#: Mode that we use for default serialization.
_JSON_MODE: Final = 'json'
# pydantic does not allow to import this,
# so we have to duplicate this type.
_IncEx: TypeAlias = (
set[int]
| set[str]
| Mapping[int, '_IncEx | bool']
| Mapping[str, '_IncEx | bool']
)
[docs]
@final
class ToJsonKwargs(TypedDict, total=False, closed=True):
"""Keyword arguments for pydantic's model dump method."""
# `mode` is explicitly left out. It is always `json`.
include: _IncEx | None
exclude: _IncEx | None
context: Any | None
by_alias: bool | None
exclude_unset: bool
exclude_defaults: bool
exclude_none: bool
exclude_computed_fields: bool
round_trip: bool
warnings: bool | Literal['none', 'warn', 'error']
fallback: Callable[[Any], Any] | None
serialize_as_any: bool
[docs]
@final
class ToModelKwargs(TypedDict, total=False, closed=True):
"""Keyword arguments for pydantic's python object validation method."""
# `from_attributes` is explicitly left out. It is always `False`.
extra: ExtraValues | None
context: Any | None
experimental_allow_partial: bool | Literal['off', 'on', 'trailing-strings']
by_alias: bool | None
by_name: bool | None
[docs]
class PydanticEndpointOptimizer(BaseEndpointOptimizer):
"""Optimize endpoints that are parsed with pydantic."""
__slots__ = ()
[docs]
@override
@classmethod
def optimize_endpoint(cls, metadata: 'EndpointMetadata') -> None:
"""Create models for return types for validation."""
# Just build all `TypeAdapter` instances
# during import time and cache them for later use in runtime.
for response in metadata.responses.values():
_get_cached_type_adapter(response.return_type) # type: ignore[arg-type]
# It is used in many places:
_get_cached_type_adapter(Any)
[docs]
class PydanticSerializer(BaseSerializer):
"""
Serialize and deserialize objects using pydantic.
Pydantic support is optional.
To install it run:
.. code:: bash
pip install 'django-modern-rest[pydantic]'
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 = pydantic_core.ValidationError
optimizer = PydanticEndpointOptimizer
schema_generator = PydanticSchemaGenerator
# Custom API:
to_json_kwargs: ClassVar[ToJsonKwargs] = {
'by_alias': True,
}
to_model_kwargs: ClassVar[ToModelKwargs] = {
'by_alias': True,
}
[docs]
@override
@classmethod
def serialize(
cls,
structure: Any,
*,
renderer: Renderer,
) -> bytes:
"""Convert any object to raw bytestring."""
try:
return renderer.render(
structure,
cls.serialize_hook,
)
except pydantic_core.PydanticSerializationError as exc:
raise DataRenderingError(str(exc)) from None
[docs]
@override
@classmethod
def serialize_hook(cls, to_serialize: Any) -> Any:
"""Customize how some objects are serialized into simple objects."""
if isinstance(to_serialize, pydantic.BaseModel):
return to_serialize.__pydantic_serializer__.to_python(
to_serialize,
mode=_JSON_MODE,
**cls.to_json_kwargs,
)
# We support dataclasses here, because raw `JsonRenderer`
# does not support them, however, we use them in multiple places inside.
# Or this is a pydantic field inside a `TypedDict`, `@dataclass`, etc:
if dataclasses.is_dataclass(to_serialize) or hasattr(
to_serialize,
'__get_pydantic_core_schema__',
):
return _get_cached_type_adapter(
type(to_serialize), # type: ignore[arg-type]
).dump_python(
to_serialize,
mode=_JSON_MODE,
**cls.to_json_kwargs,
)
return super().serialize_hook(to_serialize)
[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,
# Note: passing real `pydantic` model to `msgspec` parser
# is not supported, because it is SLOWER than using 2-phase parsing.
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 ``pydantic`` 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: Optional namespace to rebuild the type adapter.
Should be used when there are forward references
that pydantic cannot solve by itself.
Returns:
Structured and validated data.
Raises:
pydantic_core.ValidationError: When parsing can't be done.
.. versionchanged:: 0.13.0
Added *rebuild_namespace* parameter
was renamed to be *extra_namespace*.
"""
# At this point `_get_cached_type_adapter(model)` was already called
# during the optimizer stage, so it will be very fast to use in runtime.
adapter = _get_cached_type_adapter(model) # type: ignore[arg-type]
if extra_namespace is not None:
adapter.rebuild(_types_namespace=extra_namespace)
return adapter.validate_python(
unstructured,
strict=strict,
**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 build a :class:`typing.TypedDict` when there are no defaults,
because it is the fastest thing that ``pydantic`` can validate
and it does not need any conversion into keyword arguments.
When some fields have defaults, we build a regular
:func:`dataclasses.dataclass`, since ``TypedDict`` cannot have them.
Defaults are passed as-is, so :mod:`dataclasses` rules apply:
mutable defaults like ``[]`` or non-frozen models are not allowed.
.. versionadded:: 0.16.0
"""
if all(field.default is EMPTY for field in fields.values()):
annotations = {
field_name: field.annotation
for field_name, field in fields.items()
}
typed_dict = TypedDict( # type: ignore[misc]
name, # pyright: ignore[reportArgumentType] # pyrefly: ignore[name-mismatch]
annotations, # pyright: ignore[reportArgumentType]
total=True,
closed=True,
)
_get_cached_type_adapter(typed_dict) # prepare during import time
return ContextModel(typed_dict)
# We don't use `slots=True` here on purpose:
# `vars()` is the fastest way to unpack a dataclass instance,
# it requires an instance `__dict__` to exist:
dataclass = dataclasses.make_dataclass(
name,
context_field_tuples(fields),
)
_get_cached_type_adapter(dataclass) # prepare during import time
return ContextModel(dataclass, to_kwargs=vars)
[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 _get_cached_type_adapter(Any).dump_python(
structured,
mode=_JSON_MODE,
**cls.to_json_kwargs,
)
[docs]
@override
@classmethod
def serialize_validation_error(
cls,
exc: Exception,
) -> list[ErrorDetail]:
"""Serialize validation error."""
if isinstance(exc, pydantic.ValidationError):
return [
{
'msg': error['msg'],
'loc': [str(loc) for loc in error['loc']],
'type': str(ErrorType.value_error),
}
for error in exc.errors(
include_url=False,
include_context=False,
include_input=False,
)
]
raise NotImplementedError(
f'Cannot serialize exception {exc!r} of type {type(exc)} safely',
)
[docs]
class PydanticFastSerializer(PydanticSerializer):
"""
Fast pydantic serializer for cases when you only work with json.
Does not use ``parser`` and ``renderer`` passed objects, does not use
``dmr.plugins.pyndatic.PydanticSerializer.serialize_hook`` and
``dmr.plugins.pyndatic.PydanticSerializer.deserialize_hook``
method.
Is built for optimizations only, use with caution.
Only works with ``application/json`` content type.
.. versionadded:: 0.6.0
See :issue:`830`.
"""
__slots__ = ()
[docs]
@classmethod
@override
def serialize(cls, structure: Any, *, renderer: Renderer) -> bytes:
"""
Fast way to serializer pyndatic models into json bytestring.
*renderer* parameter is always ignored.
"""
try:
return _get_cached_type_adapter(
type(structure), # type: ignore[arg-type]
).dump_json(
structure,
fallback=cls.serialize_hook,
**cls.to_json_kwargs, # type: ignore[misc]
)
except pydantic_core.PydanticSerializationError as exc:
raise DataRenderingError(str(exc)) from exc
[docs]
@classmethod
@override
def deserialize(
cls,
buffer: Raw,
*,
parser: Parser,
request: HttpRequest,
model: TypeForm[Any],
) -> Any:
"""
Fast way to serializer pyndatic models into json bytestring.
*parser* parameter is always ignored.
.. versionchanged:: 0.16.0
Now uses faster ``json_loads`` when it is available.
"""
try:
return _json_loads(buffer)
except pydantic_core.ValidationError as exc:
# Corner case: an empty body is `None` for us,
# just like `JsonParser` treats it. Happens for `204` responses.
# We do this here, because we don't want
# a penalty for all positive cases.
if buffer == b'':
return None
raise DataParsingError(exc.errors()[0]['msg']) from exc
except Exception as exc: # which can be raise for non-pydantic loads
if buffer == b'':
return None
raise DataParsingError(str(exc)) from exc
[docs]
@classmethod
@override
def validate(
cls,
controller_cls: type['Controller[BaseSerializer]'],
metadata: 'EndpointMetadata',
) -> None:
"""
Validate that only ``json`` parsers and renderers are used.
.. versionchanged:: 0.16.0
Replaces ``is_supported`` method.
"""
pluggables: list[Parser | Renderer] = [
*metadata.parsers.values(),
*metadata.renderers.values(),
]
for pluggable in pluggables:
if pluggable.content_type != 'application/json':
raise EndpointMetadataError(
f'{metadata.endpoint_name!r} serializer '
f'does not support {pluggable!r}, only json is supported',
)
_ModelT = TypeVar('_ModelT')
@lru_cache(maxsize=MAX_CACHE_SIZE)
def _get_cached_type_adapter(model: _ModelT) -> pydantic.TypeAdapter[_ModelT]:
"""
It is expensive to create, reuse existing ones.
If you want to clear this cache run:
.. code:: python
>>> _get_cached_type_adapter.cache_clear()
Or use :func:`dmr.settings.clear_settings_cache`.
"""
# This is a function not to cache `self` or `cls` params.
return pydantic.TypeAdapter(model, _parent_depth=4)
#: Internal helper to load json for fast serializer.
#: If `msgspec` is missing and the default `json` module is used, we fallback
#: to the `pydantic` one, which is also rather slow.
_json_loads: Final = (
_get_cached_type_adapter(Any).validate_json
if json_loads is NativeJson.loads
else json_loads
)
django-modern-rest