Implement basic abstractions around pydantic 2.

.editorconfig

@@ -0,0 +1,16 @@
+root = true
+
+[*]
+charset = utf-8
+end_of_line = lf
+insert_final_newline = true
+trim_trailing_whitespace = true
+indent_style = space
+indent_size = 4
+
+[*.py]
+max_line_length = 120
+
+[{*.json,*.yml,*.yaml}]
+indent_size = 2
+insert_final_newline = false

django_pydantic_field/v2/__init__.py

@@ -0,0 +1 @@
+from .fields import SchemaField as SchemaField

django_pydantic_field/v2/fields.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import typing as ty
+
+import pydantic
+
+from django.core import checks, exceptions
+
+from django.db.models.expressions import BaseExpression
+from django.db.models.fields.json import JSONField
+from django.db.models.query_utils import DeferredAttribute
+
+from . import types
+
+if ty.TYPE_CHECKING:
+    from ..compat.django import GenericContainer
+
+
+class SchemaAttribute(DeferredAttribute):
+    field: PydanticSchemaField
+
+    def __set_name__(self, owner, name):
+        self.field.adapter.bind(owner, name)
+
+    def __set__(self, obj, value):
+        obj.__dict__[self.field.attname] = self.field.to_python(value)
+
+
+class PydanticSchemaField(JSONField, ty.Generic[types.ST]):
+    def __init__(
+        self,
+        *args,
+        schema: type[types.ST] | GenericContainer | ty.ForwardRef | str | None = None,
+        config: pydantic.ConfigDict | None = None,
+        **kwargs,
+    ):
+        self.export_kwargs = export_kwargs = types.SchemaAdapter.extract_export_kwargs(kwargs)
+        super().__init__(*args, **kwargs)
+        self.schema = schema
+        self.config = config
+        self.adapter = types.SchemaAdapter(schema, config, None, self.get_attname(), self.null, **export_kwargs)
+
+    def __copy__(self):
+        _, _, args, kwargs = self.deconstruct()
+        copied = self.__class__(*args, **kwargs)
+        copied.set_attributes_from_name(self.name)
+        return copied
+
+    def deconstruct(self) -> ty.Any:
+        field_name, import_path, args, kwargs = super().deconstruct()
+        kwargs.update(schema=GenericContainer.wrap(self.schema), config=self.config, **self.export_kwargs)
+        return field_name, import_path, args, kwargs
+
+    def contribute_to_class(self, cls: types.DjangoModelType, name: str, private_only: bool = False) -> None:
+        self.adapter.bind(cls, name)
+        super().contribute_to_class(cls, name, private_only)
+
+    def check(self, **kwargs: ty.Any) -> list[checks.CheckMessage]:
+        performed_checks = super().check(**kwargs)
+        try:
+            self.adapter.validate_schema()
+        except ValueError as exc:
+            performed_checks.append(checks.Error(exc.args[0], obj=self))
+        return performed_checks
+
+    def to_python(self, value: ty.Any):
+        try:
+            return self.adapter.validate_python(value)
+        except pydantic.ValidationError as exc:
+            raise exceptions.ValidationError(exc.title, code="invalid", params=exc.errors()) from exc
+
+    def get_prep_value(self, value: ty.Any):
+        if isinstance(value, BaseExpression):
+            # We don't want to perform coercion on database query expressions.
+            return super().get_prep_value(value)
+        return self.adapter.dump_python(value)
+
+    def validate(self, value: ty.Any, model_instance: ty.Any) -> None:
+        value = self.adapter.validate_python(value)
+        return super().validate(value, model_instance)
+
+
+@ty.overload
+def SchemaField(schema: None = None) -> ty.Any:
+    ...
+
+
+@ty.overload
+def SchemaField(schema: type[types.ST]) -> ty.Any:
+    ...
+
+
+def SchemaField(schema=None, config=None, *args, **kwargs):  # type: ignore
+    return PydanticSchemaField(*args, schema=schema, config=config, **kwargs)

django_pydantic_field/v2/types.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+import functools
+import typing as ty
+
+from pydantic.type_adapter import TypeAdapter
+
+from . import utils
+from ..compat.django import GenericContainer
+
+ST = ty.TypeVar("ST", bound="SchemaT")
+
+if ty.TYPE_CHECKING:
+    from pydantic import BaseModel
+    from pydantic.type_adapter import IncEx
+    from pydantic.dataclasses import DataclassClassOrWrapper
+    from django.db.models import Model
+
+    ModelType = ty.Type[BaseModel]
+    DjangoModelType = ty.Type[Model]
+    SchemaT = ty.Union[
+        BaseModel,
+        DataclassClassOrWrapper,
+        ty.Sequence[ty.Any],
+        ty.Mapping[str, ty.Any],
+        ty.Set[ty.Any],
+        ty.FrozenSet[ty.Any],
+    ]
+
+class ExportKwargs(ty.TypedDict, total=False):
+    strict: bool
+    mode: ty.Literal["json", "python"]
+    include: IncEx | None
+    exclude: IncEx | None
+    by_alias: bool
+    exclude_unset: bool
+    exclude_defaults: bool
+    exclude_none: bool
+    round_trip: bool
+    warnings: bool
+
+
+class SchemaAdapter(ty.Generic[ST]):
+    def __init__(
+        self,
+        schema,
+        config,
+        parent_type,
+        attname,
+        allow_null,
+        *,
+        parent_depth=4,
+        **export_kwargs: ty.Unpack[ExportKwargs],
+    ):
+        self.schema = schema
+        self.config = config
+        self.parent_type = parent_type
+        self.attname = attname
+        self.allow_null = allow_null
+        self.parent_depth = parent_depth
+        self.export_kwargs = export_kwargs
+
+    @staticmethod
+    def extract_export_kwargs(kwargs: dict[str, ty.Any]) -> ExportKwargs:
+        common_keys = kwargs.keys() & ExportKwargs.__annotations__.keys()
+        export_kwargs = {key: kwargs.pop(key) for key in common_keys}
+        return ty.cast(ExportKwargs, export_kwargs)
+
+    @functools.cached_property
+    def type_adapter(self) -> TypeAdapter:
+        schema = self._get_prepared_schema()
+        return TypeAdapter(schema, config=self.config, _parent_depth=4)  # type: ignore
+
+    def bind(self, parent_type, attname):
+        self.parent_type = parent_type
+        self.attname = attname
+        self.__dict__.pop("type_adapter", None)
+
+    def validate_schema(self) -> None:
+        """Validate the schema and raise an exception if it is invalid."""
+        self._get_prepared_schema()
+
+    def validate_python(self, value: ty.Any) -> ST:
+        """Validate the value and raise an exception if it is invalid."""
+        return self.type_adapter.validate_python(
+            value,
+            strict=self.export_kwargs.get("strict", None),
+        )
+
+    def dump_python(self, value: ty.Any) -> ty.Any:
+        """Dump the value to a Python object."""
+        return self.type_adapter.dump_python(value, **self._dump_python_kwargs)
+
+    def json_schema(self) -> ty.Any:
+        """Return the JSON schema for the field."""
+        by_alias = self.export_kwargs.get("by_alias", True)
+        return self.type_adapter.json_schema(by_alias=by_alias)
+
+    def _get_prepared_schema(self) -> type[ST]:
+        schema = self.schema
+
+        if schema is None:
+            schema = self._guess_schema_from_annotations()
+        if isinstance(schema, GenericContainer):
+            schema = ty.cast(type[ST], GenericContainer.unwrap(schema))
+        if isinstance(schema, (str, ty.ForwardRef)):
+            schema = self._resolve_schema_forward_ref(schema)
+
+        if schema is None:
+            error_msg = f"Schema not provided for {self.parent_type.__name__}.{self.attname}"
+            raise ValueError(error_msg)
+
+        if self.allow_null:
+            schema = ty.Optional[schema]
+        return ty.cast(type[ST], schema)
+
+    def _guess_schema_from_annotations(self) -> type[ST] | str | ty.ForwardRef | None:
+        return utils.get_annotated_type(self.parent_type, self.attname)
+
+    def _resolve_schema_forward_ref(self, schema: str | ty.ForwardRef) -> ty.Any:
+        if isinstance(schema, str):
+            schema = ty.ForwardRef(schema)
+        namespace = utils.get_local_namespace(self.parent_type)
+        return schema._evaluate(namespace, vars(self.parent_type), frozenset())  # type: ignore
+
+    @functools.cached_property
+    def _dump_python_kwargs(self) -> dict[str, ty.Any]:
+        export_kwargs = self.export_kwargs.copy()
+        export_kwargs.pop("strict", None)
+        return ty.cast(dict, export_kwargs)

django_pydantic_field/v2/utils.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import sys
+import typing as ty
+
+
+def get_annotated_type(obj, field, default=None) -> ty.Any:
+    try:
+        if isinstance(obj, type):
+            annotations = obj.__dict__["__annotations__"]
+        else:
+            annotations = obj.__annotations__
+
+        return annotations[field]
+    except (AttributeError, KeyError):
+        return default
+
+
+def get_local_namespace(cls) -> dict[str, ty.Any]:
+    try:
+        module = cls.__module__
+        return vars(sys.modules[module])
+    except (KeyError, AttributeError):
+        return {}

pyproject.toml

@@ -75,6 +75,31 @@ Changelog = "https://github.com/surenkov/django-pydantic-field/releases"
 [tool.setuptools]
 packages = ["django_pydantic_field"]
 
+[tool.isort]
+py_version = 312
+profile = "black"
+line_length = 120
+multi_line_output = 3
+include_trailing_comma = true
+force_alphabetical_sort_within_sections = true
+force_grid_wrap = 0
+use_parentheses = true
+skip_glob = [
+    "*/migrations/*",
+]
+
+[tool.black]
+target-version = ['py312']
+line-length = 120
+exclude = '''
+/(
+    \.pytest_cache
+  | \.venv
+  | venv
+  | migrations
+)/
+'''
+
 [tool.mypy]
 plugins = [
     "mypy_django_plugin.main",