from collections.abc import Callable, Mapping
from dataclasses import is_dataclass
from functools import lru_cache
from typing import (
TYPE_CHECKING,
Any,
ClassVar,
Literal,
TypeAlias,
Union,
final,
)
import pydantic
import pydantic_core
from django.http import HttpRequest
from pydantic.config import ExtraValues
from typing_extensions import TypedDict, override
from dmr.envs import MAX_CACHE_SIZE
from dmr.errors import ErrorDetail, ErrorType
from dmr.exceptions import DataRenderingError
from dmr.parsers import Parser, Raw
from dmr.plugins.pydantic.schema import PydanticSchemaGenerator
from dmr.renderers import Renderer
from dmr.serializer import BaseEndpointOptimizer, BaseSerializer
if TYPE_CHECKING:
from dmr.metadata import EndpointMetadata
# pydantic does not allow to import this,
# so we have to duplicate this type.
_IncEx: TypeAlias = (
set[int]
| set[str]
| Mapping[int, Union['_IncEx', bool]]
| Mapping[str, Union['_IncEx', bool]]
)
@final
class ModelDumpKwargs(TypedDict, total=False):
"""Keyword arguments for pydantic's model dump method."""
mode: Literal['json', 'python'] | str
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
@final
class FromPythonKwargs(TypedDict, total=False):
"""Keyword arguments for pydantic's python object validation method."""
extra: ExtraValues | None
from_attributes: bool | 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."""
[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)
[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]'
"""
__slots__ = ()
# Required API:
validation_error = pydantic_core.ValidationError
optimizer = PydanticEndpointOptimizer
schema_generator = PydanticSchemaGenerator
# Custom API:
model_dump_kwargs: ClassVar[ModelDumpKwargs] = {
'by_alias': True,
'mode': 'json',
}
from_python_kwargs: ClassVar[FromPythonKwargs] = {
'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.model_dump(**cls.model_dump_kwargs)
if is_dataclass(to_serialize):
return _get_cached_type_adapter(
type(to_serialize), # type: ignore[arg-type]
).dump_python(
to_serialize,
)
return super().serialize_hook(to_serialize)
[docs]
@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,
)
[docs]
@override
@classmethod
def from_python(
cls,
unstructured: Any,
model: Any,
*,
strict: bool | None,
rebuild_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.
rebuild_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.
"""
# 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)
if rebuild_namespace is not None:
adapter.rebuild(_types_namespace=rebuild_namespace)
return adapter.validate_python(
unstructured,
strict=strict,
**cls.from_python_kwargs,
)
[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,
# To be in sync with `msgspec`:
mode='json',
)
[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',
)
@lru_cache(maxsize=MAX_CACHE_SIZE)
def _get_cached_type_adapter(model: Any) -> pydantic.TypeAdapter[Any]:
"""
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)
django-modern-rest