Support typing.Annotated annotations (#52)

This PR is related to #31

This PR adds the support of typing.Annotated[...] expressions, both through schema= attribute or field annotation syntax; though I find the schema=typing.Annotated[...] variant highly discouraged.

The current limitation is not in the field itself, but in possible Annotated metadata -- practically it can contain anything, and Django migrations serializers could refuse to write it to migrations.

For most relevant types in context of Pydantic, I wrote the specific serializers (particularly for pydantic.FieldInfo, pydantic.Representation and raw dataclasses), thus it should cover the majority of Annotated use cases

.pre-commit-config.yaml

@@ -11,11 +11,11 @@ repos:
       - id: isort
         args: [ "--settings-path", "./pyproject.toml", "--filter-files" ]
         files: "^django_pydantic_field/"
-        exclude: ^.*\b(\.pytest_cache|\.venv|venv|tests)\b.*$
+        exclude: ^.*\b(\.pytest_cache|\.venv|venv).*\b.*$
   - repo: https://github.com/psf/black
     rev: 24.3.0
     hooks:
       - id: black
         args: [ "--config", "./pyproject.toml" ]
         files: "^django_pydantic_field/"
-        exclude: ^.*\b(\.pytest_cache|\.venv|venv|tests)\b.*$
+        exclude: ^.*\b(\.pytest_cache|\.venv|venv).*\b.*$

README.md

@@ -82,8 +82,21 @@ class Bar(pydantic.BaseModel):
     slug: str = "foo_bar"
 ```
 
-In this case, exact type resolution will be postponed until initial access to the field.
-Usually this happens on the first instantiation of the model.
+**Pydantic v2 specific**: this behaviour is achieved by the fact that the exact type resolution will be postponed the until initial access to the field. Usually this happens on the first instantiation of the model.
+
+To reduce the number of runtime errors related to the postponed resolution, the field itself performs a few checks against the passed schema during `./manage.py check` command invocation, and consequently, in `runserver` and `makemigrations` commands.
+
+Here's the list of currently implemented checks:
+- `pydantic.E001`: The passed schema could not be resolved. Most likely it does not exist in the scope of the defined field.
+- `pydantic.E002`: `default=` value could not be serialized to the schema.
+- `pydantic.W003`: The default value could not be reconstructed to the schema due to `include`/`exclude` configuration.
+
+
+### `typing.Annotated` support
+As of `v0.3.5`, SchemaField also supports `typing.Annotated[...]` expressions, both through `schema=` attribute or field annotation syntax; though I find the `schema=typing.Annotated[...]` variant highly discouraged.
+
+**The current limitation** is not in the field itself, but in possible `Annotated` metadata -- practically it can contain anything, and Django migrations serializers could refuse to write it to migrations.
+For most relevant types in context of Pydantic, I wrote the specific serializers (particularly for `pydantic.FieldInfo`, `pydantic.Representation` and raw dataclasses), thus it should cover the majority of `Annotated` use cases.
 
 ## Django Forms support
 

django_pydantic_field/compat/django.py

@@ -13,87 +13,252 @@
     `typing.Union` and its special forms, like `typing.Optional`, have its own inheritance chain.
     Moreover, `types.UnionType`, introduced in 3.10, do not allow explicit type construction,
     only with `X | Y` syntax. Both cases require a dedicated serializer for migration writes.
+
+[typing.Annotated](https://peps.python.org/pep-0593/)
+    `typing.Annotated` syntax is supported for direct field annotations, though I find it highly discouraged
+    while using in `schema=` attribute.
+    The limitation with `Annotated` types is that supplied metadata could be actually of any type.
+    In case of Pydantic, it is a `FieldInfo` objects, which are not compatible with Django Migrations serializers.
+    This module provides a few containers (`FieldInfoContainer` and `DataclassContainer`),
+    which allow Model serializers to work.
 """
 
+from __future__ import annotations
+
+import abc
+import dataclasses
 import sys
 import types
 import typing as ty
 
+import typing_extensions as te
 from django.db.migrations.serializer import BaseSerializer, serializer_factory
 from django.db.migrations.writer import MigrationWriter
+from pydantic.fields import FieldInfo
+from pydantic_core import PydanticUndefined
 
+from .pydantic import PYDANTIC_V1
 from .typing import get_args, get_origin
 
+try:
+    from pydantic._internal._repr import Representation
+    from pydantic.fields import _DefaultValues as FieldInfoDefaultValues
+    from pydantic_core import PydanticUndefined
+except ImportError:
+    # Assuming this is a Pydantic v1
+    from pydantic.fields import Undefined as PydanticUndefined  # type: ignore[attr-defined, no-redef]
+    from pydantic.utils import Representation  # type: ignore[no-redef]
+
+    FieldInfoDefaultValues = FieldInfo.__field_constraints__  # type: ignore[attr-defined]
+
+
+class BaseContainer(abc.ABC):
+    __slot__ = ()
 
-class GenericContainer:
+    @classmethod
+    def unwrap(cls, value):
+        if isinstance(value, BaseContainer) and type(value) is not BaseContainer:
+            return value.unwrap(value)
+        return value
+
+    def __eq__(self, other):
+        if isinstance(other, self.__class__):
+            return all(getattr(self, attr) == getattr(other, attr) for attr in self.__slots__)
+        return NotImplemented
+
+    def __str__(self):
+        return repr(self.unwrap(self))
+
+    def __repr__(self):
+        attrs = tuple(getattr(self, attr) for attr in self.__slots__)
+        return f"{self.__class__.__name__}{attrs}"
+
+
+class GenericContainer(BaseContainer):
     __slots__ = "origin", "args"
 
     def __init__(self, origin, args: tuple = ()):
         self.origin = origin
         self.args = args
 
     @classmethod
-    def wrap(cls, typ_):
-        if isinstance(typ_, GenericTypes):
-            wrapped_args = tuple(map(cls.wrap, get_args(typ_)))
-            return cls(get_origin(typ_), wrapped_args)
-        return typ_
+    def wrap(cls, value):
+        # NOTE: due to a bug in typing_extensions for `3.8`, Annotated aliases are handled explicitly
+        if isinstance(value, AnnotatedAlias):
+            args = (value.__origin__, *value.__metadata__)
+            wrapped_args = tuple(map(cls.wrap, args))
+            return cls(te.Annotated, wrapped_args)
+        if isinstance(value, GenericTypes):
+            wrapped_args = tuple(map(cls.wrap, get_args(value)))
+            return cls(get_origin(value), wrapped_args)
+        if isinstance(value, FieldInfo):
+            return FieldInfoContainer.wrap(value)
+        return value
 
     @classmethod
-    def unwrap(cls, type_):
-        if not isinstance(type_, cls):
-            return type_
+    def unwrap(cls, value):
+        if not isinstance(value, cls):
+            return value
+
+        if PYDANTIC_V1:
+            origin = get_origin(BaseContainer.unwrap(value.origin)) or value.origin
+        else:
+            origin = value.origin
 
-        if not type_.args:
-            return type_.origin
+        if not value.args:
+            return origin
 
-        unwrapped_args = tuple(map(cls.unwrap, type_.args))
+        unwrapped_args = tuple(map(BaseContainer.unwrap, value.args))
         try:
             # This is a fallback for Python < 3.8, please be careful with that
-            return type_.origin[unwrapped_args]
+            return origin[unwrapped_args]
         except TypeError:
-            return GenericAlias(type_.origin, unwrapped_args)
+            return GenericAlias(origin, unwrapped_args)
+
+    def __eq__(self, other):
+        if isinstance(other, GenericTypes):
+            return self == self.wrap(other)
+        return super().__eq__(other)
 
-    def __repr__(self):
-        return repr(self.unwrap(self))
 
-    __str__ = __repr__
+class DataclassContainer(BaseContainer):
+    __slots__ = "datacls", "kwargs"
+
+    def __init__(self, datacls: type, kwargs: ty.Dict[str, ty.Any]):
+        self.datacls = datacls
+        self.kwargs = kwargs
+
+    @classmethod
+    def wrap(cls, value):
+        if cls._is_dataclass_instance(value):
+            return cls(type(value), dataclasses.asdict(value))
+        if isinstance(value, GenericTypes):
+            return GenericContainer.wrap(value)
+        return value
+
+    @classmethod
+    def unwrap(cls, value):
+        if isinstance(value, cls):
+            return value.datacls(**value.kwargs)
+        return value
+
+    @staticmethod
+    def _is_dataclass_instance(obj: ty.Any):
+        return dataclasses.is_dataclass(obj) and not isinstance(obj, type)
 
     def __eq__(self, other):
-        if isinstance(other, self.__class__):
-            return self.origin == other.origin and self.args == other.args
-        if isinstance(other, GenericTypes):
+        if self._is_dataclass_instance(other):
             return self == self.wrap(other)
-        return NotImplemented
+        return super().__eq__(other)
+
+
+class FieldInfoContainer(BaseContainer):
+    __slots__ = "origin", "metadata", "kwargs"
+
+    def __init__(self, origin, metadata, kwargs):
+        self.origin = origin
+        self.metadata = metadata
+        self.kwargs = kwargs
+
+    @classmethod
+    def wrap(cls, field: FieldInfo):
+        if not isinstance(field, FieldInfo):
+            return field
+
+        # `getattr` is important to preserve compatibility with Pydantic v1
+        metadata = getattr(field, "metadata", ())
+        origin = getattr(field, "annotation", None)
+        if origin is type(None):
+            origin = None
+
+        origin = GenericContainer.wrap(origin)
+        metadata = tuple(map(DataclassContainer.wrap, metadata))
+
+        kwargs = dict(cls._iter_field_attrs(field))
+        if PYDANTIC_V1:
+            kwargs.update(kwargs.pop("extra", {}))
+
+        return cls(origin, metadata, kwargs)
 
+    @classmethod
+    def unwrap(cls, value):
+        if not isinstance(value, cls):
+            return value
+        if PYDANTIC_V1:
+            return FieldInfo(**value.kwargs)
+
+        origin = GenericContainer.unwrap(value.origin)
+        metadata = tuple(map(BaseContainer.unwrap, value.metadata))
+        try:
+            annotated_args = (origin, *metadata)
+            annotation = te.Annotated[annotated_args]
+        except TypeError:
+            annotation = None
 
-class GenericSerializer(BaseSerializer):
-    value: GenericContainer
+        return FieldInfo(annotation=annotation, **value.kwargs)
+
+    def __eq__(self, other):
+        if isinstance(other, FieldInfo):
+            return self == self.wrap(other)
+        return super().__eq__(other)
+
+    @staticmethod
+    def _iter_field_attrs(field: FieldInfo):
+        available_attrs = set(field.__slots__) - {"annotation", "metadata", "_attributes_set"}
+
+        for attr in available_attrs:
+            attr_value = getattr(field, attr)
+            if attr_value is not PydanticUndefined and attr_value != FieldInfoDefaultValues.get(attr):
+                yield attr, getattr(field, attr)
+
+    @staticmethod
+    def _wrap_metadata_object(obj):
+        return DataclassContainer.wrap(obj)
+
+
+class BaseContainerSerializer(BaseSerializer):
+    value: BaseContainer
 
     def serialize(self):
-        value = self.value
+        tp_repr, imports = serializer_factory(type(self.value)).serialize()
+        attrs = []
 
-        tp_repr, imports = serializer_factory(type(value)).serialize()
-        orig_repr, orig_imports = serializer_factory(value.origin).serialize()
-        imports.update(orig_imports)
+        for attr in self._iter_container_attrs():
+            attr_repr, attr_imports = serializer_factory(attr).serialize()
+            attrs.append(attr_repr)
+            imports.update(attr_imports)
 
-        args = []
-        for arg in value.args:
-            arg_repr, arg_imports = serializer_factory(arg).serialize()
-            args.append(arg_repr)
-            imports.update(arg_imports)
+        attrs_repr = ", ".join(attrs)
+        return f"{tp_repr}({attrs_repr})", imports
 
-        if args:
-            args_repr = ", ".join(args)
-            generic_repr = "%s(%s, (%s,))" % (tp_repr, orig_repr, args_repr)
-        else:
-            generic_repr = "%s(%s)" % (tp_repr, orig_repr)
+    def _iter_container_attrs(self):
+        container = self.value
+        for attr in container.__slots__:
+            yield getattr(container, attr)
+
+
+class DataclassContainerSerializer(BaseSerializer):
+    value: DataclassContainer
 
-        return generic_repr, imports
+    def serialize(self):
+        tp_repr, imports = serializer_factory(self.value.datacls).serialize()
+
+        kwarg_pairs = []
+        for arg, value in self.value.kwargs.items():
+            value_repr, value_imports = serializer_factory(value).serialize()
+            kwarg_pairs.append(f"{arg}={value_repr}")
+            imports.update(value_imports)
+
+        kwargs_repr = ", ".join(kwarg_pairs)
+        return f"{tp_repr}({kwargs_repr})", imports
 
 
 class TypingSerializer(BaseSerializer):
     def serialize(self):
+        value = GenericContainer.wrap(self.value)
+        if isinstance(value, GenericContainer):
+            return serializer_factory(value).serialize()
+
         orig_module = self.value.__module__
         orig_repr = repr(self.value)
 
@@ -103,6 +268,36 @@ def serialize(self):
         return orig_repr, {f"import {orig_module}"}
 
 
+class FieldInfoSerializer(BaseSerializer):
+    value: FieldInfo
+
+    def serialize(self):
+        container = FieldInfoContainer.wrap(self.value)
+        return serializer_factory(container).serialize()
+
+
+class RepresentationSerializer(BaseSerializer):
+    value: Representation
+
+    def serialize(self):
+        tp_repr, imports = serializer_factory(type(self.value)).serialize()
+        repr_args = []
+
+        for arg_name, arg_value in self.value.__repr_args__():
+            arg_value_repr, arg_value_imports = serializer_factory(arg_value).serialize()
+            imports.update(arg_value_imports)
+
+            if arg_name is None:
+                repr_args.append(arg_value_repr)
+            else:
+                repr_args.append(f"{arg_name}={arg_value_repr}")
+
+        final_args_repr = ", ".join(repr_args)
+        return f"{tp_repr}({final_args_repr})"
+
+
+AnnotatedAlias = te._AnnotatedAlias
+
 if sys.version_info >= (3, 9):
     GenericAlias = types.GenericAlias
     GenericTypes: ty.Tuple[ty.Any, ...] = (
@@ -117,7 +312,18 @@ def serialize(self):
     GenericTypes = GenericAlias, type(ty.List)  # noqa
 
 
-MigrationWriter.register_serializer(GenericContainer, GenericSerializer)
+# BaseContainerSerializer *must be* registered after all specialized container serializers
+MigrationWriter.register_serializer(DataclassContainer, DataclassContainerSerializer)
+MigrationWriter.register_serializer(BaseContainer, BaseContainerSerializer)
+
+# Pydantic-specific datastructures serializers
+MigrationWriter.register_serializer(FieldInfo, FieldInfoSerializer)
+MigrationWriter.register_serializer(Representation, RepresentationSerializer)
+
+# Typing serializers
+for type_ in GenericTypes:
+    MigrationWriter.register_serializer(type_, TypingSerializer)
+
 MigrationWriter.register_serializer(ty.ForwardRef, TypingSerializer)
 MigrationWriter.register_serializer(type(ty.Union), TypingSerializer)  # type: ignore