pydantic 2.13

.pre-commit-config.yaml

@@ -44,7 +44,7 @@ repos:
         args: [--py39-plus, --keep-runtime-typing]
 
   - repo: https://github.com/adrienverge/yamllint.git
-    rev: v1.37.1
+    rev: v1.38.0
     hooks:
       - id: yamllint
         args: ["-d", "{extends: relaxed, rules: {empty-lines: disable, line-length: {max: 1500}}}", --strict, --format, parsable]
@@ -58,7 +58,7 @@ repos:
 
   - repo: https://github.com/astral-sh/uv-pre-commit
     # uv version.
-    rev: 0.9.13
+    rev: 0.9.29
     hooks:
       # Update the uv lockfile
       - id: uv-lock

pyproject.toml

@@ -6,11 +6,12 @@ authors = [
 ]
 dependencies = [
     "PyYaml",
-    "pydantic == 2.11.9",
+    "pydantic @ git+https://github.com/pydantic/pydantic.git",
     "email-validator",
     "yarl",
     "httpx",
     "more-itertools",
+    'typing_extensions; python_version<"3.10"',
     "jmespath",
 ]
 requires-python = ">=3.10"
@@ -102,6 +103,7 @@ filterwarnings = [
     "ignore:'flask.Markup' is deprecated and will be removed in Flask 2.4. Import 'markupsafe.Markup' instead.:DeprecationWarning",
     "ignore:unclosed resource <TCPTransport:ResourceWarning",
     "ignore:co_lnotab is deprecated, use co_lines instead:DeprecationWarning",
+#    "ignore::pydantic.warnings.UnsupportedFieldAttributeWarning",
 ]
 asyncio_mode = "strict"
 asyncio_default_fixture_loop_scope = "session"

requirements.txt

@@ -35,9 +35,9 @@ multidict==6.7.0
     # via yarl
 propcache==0.4.1
     # via yarl
-pydantic==2.11.9
+pydantic @ git+https://github.com/pydantic/pydantic.git@c42224acb31dd9f374124a74809332246e595194
     # via aiopenapi3
-pydantic-core==2.33.2
+pydantic-core @ git+https://github.com/pydantic/pydantic.git@c42224acb31dd9f374124a74809332246e595194#subdirectory=pydantic-core
     # via pydantic
 pyyaml==6.0.3
     # via aiopenapi3

src/aiopenapi3/model.py

@@ -250,13 +250,13 @@ def model(self) -> type[BaseModel] | type[None]:
         return m
 
     @classmethod
-    def collapse(cls, type_name, items: list["_ClassInfo"]) -> type[BaseModel]:
+    def collapse(cls, schema: "SchemaType", items: list["_ClassInfo"]) -> type[BaseModel]:
         r: list[type[BaseModel] | type[None]]
-
         r = [i.model() for i in items]
+        type_name = schema._get_identity("L8")
 
         if len(r) > 1:
-            ru: object = Union[tuple(r)]
+            ru = Annotated[Union[tuple(r)], Field(default=getattr(schema, "default", None))]
             m: type[RootModel] = create_model(type_name, __base__=(ConfiguredRootModel[ru],), __module__=me.__name__)
         elif len(r) == 1:
             m: type[BaseModel] = cast(type[BaseModel], r[0])
@@ -297,10 +297,17 @@ def from_schema(
 
         r: list[_ClassInfo] = list()
 
-        for _type in Model.types(schema):
-            r.append(Model.createClassInfo(schema, _type, schemanames, discriminators, extra))
+        types: list[str] = list(Model.types(schema))
+        multi: bool = len(types) > 1
+        for _type in types:
+            args = dict() if multi else None
+            """
+            for schema with multiple types, the default value needs to be attached to the RootModel
+            providing empty args creates a FieldInfo without a default value for the subtypes
+            """
+            r.append(Model.createClassInfo(schema, _type, schemanames, discriminators, extra, args))
 
-        m = _ClassInfo.collapse(schema._get_identity("L8"), r)
+        m = _ClassInfo.collapse(schema, r)
 
         return cast(type[BaseModel], m)
 
@@ -312,6 +319,7 @@ def createClassInfo(
         schemanames: list[str],
         discriminators: list["DiscriminatorType"],
         extra: list["SchemaType"] | None,
+        args: dict[str, Any] = None,
     ) -> _ClassInfo:
         from . import v20, v30, v31
 
@@ -325,9 +333,10 @@ def createClassInfo(
             for primitive types the anyOf/oneOf is taken care of in Model.createAnnotation
             """
             if typing.get_origin(_t := Model.createAnnotation(schema, _type=_type)) != Literal:
-                classinfo.root = Annotated[_t, Model.createField(schema, _type=_type, args=None)]
+                classinfo.root = Annotated[_t, Model.createField(schema, _type=_type, args=args)]
             else:
                 classinfo.root = _t
+
         elif _type == "array":
             """anyOf/oneOf is taken care in in createAnnotation"""
             classinfo.root = Model.createAnnotation(schema, _type="array")
@@ -608,7 +617,7 @@ def createAnnotation(
         return rr
 
     @staticmethod
-    def types(schema: "SchemaType"):
+    def types(schema: "SchemaType") -> typing.Generator[str, None, None]:
         if isinstance(schema.type, str):
             yield schema.type
             if getattr(schema, "nullable", False):
@@ -713,7 +722,7 @@ def booleanFalse(schema: Optional["SchemaType"]) -> bool:
             raise ValueError(schema)
 
     @staticmethod
-    def createField(schema: "SchemaType", _type=None, args=None):
+    def createField(schema: "SchemaType", _type=None, args=None) -> Field:
         if args is None:
             args = dict(default=getattr(schema, "default", None))
 

src/aiopenapi3/v30/parameter.py

@@ -316,9 +316,11 @@ class Parameter(ParameterBase, _ParameterCodec):
     in_: _In = Field(alias="in")  # TODO must be one of ["query","header","path","cookie"]
 
     @model_validator(mode="after")
-    def validate_Parameter(cls, p: "ParameterBase"):
-        assert p.in_ != "path" or p.required is True, "Parameter '%s' must be required since it is in the path" % p.name
-        return p
+    def validate_Parameter(self):
+        assert self.in_ != "path" or self.required is True, (
+            "Parameter '%s' must be required since it is in the path" % self.name
+        )
+        return self
 
 
 def encode_parameter(

src/aiopenapi3/v30/paths.py

@@ -38,15 +38,14 @@ class Link(ObjectExtended):
     server: Server | None = Field(default=None)
 
     @model_validator(mode="after")
-    @classmethod
-    def validate_Link_operation(cls, l: '__types["Link"]'):  # noqa: F821
-        assert not (l.operationId is not None and l.operationRef is not None), (
+    def validate_Link_operation(self):  # type: ignore[name-defined]
+        assert not (self.operationId is not None and self.operationRef is not None), (
             "operationId and operationRef are mutually exclusive, only one of them is allowed"
         )
-        assert not (l.operationId == l.operationRef is None), (
+        assert not (self.operationId == self.operationRef is None), (
             "operationId and operationRef are mutually exclusive, one of them must be specified"
         )
-        return l
+        return self
 
 
 class Response(ObjectExtended):

src/aiopenapi3/v30/schemas.py

@@ -74,13 +74,12 @@ def is_boolean_schema(cls, data: Any) -> Any:
             return {"not": {}}
 
     @model_validator(mode="after")
-    @classmethod
-    def validate_Schema_number_type(cls, s: "Schema"):
-        if s.type == "integer":
+    def validate_Schema_number_type(self):
+        if self.type == "integer":
             for i in ["minimum", "maximum"]:
-                if (v := getattr(s, i, None)) is not None and not isinstance(v, int):
-                    setattr(s, i, int(v))
-        return s
+                if (v := getattr(self, i, None)) is not None and not isinstance(v, int):
+                    setattr(self, i, int(v))
+        return self
 
     def __getstate__(self):
         return SchemaBase.__getstate__(self)

src/aiopenapi3/v30/servers.py

@@ -17,11 +17,11 @@ class ServerVariable(ObjectExtended):
     description: str | None = Field(default=None)
 
     @model_validator(mode="after")
-    def validate_ServerVariable(cls, s: "ServerVariable"):
-        assert isinstance(s.enum, (list, None.__class__))
+    def validate_ServerVariable(self):
+        assert isinstance(self.enum, (list, None.__class__))
         # default value must be in enum
-        assert s.default is None or s.default in (s.enum or [s.default])
-        return s
+        assert self.default is None or self.default in (self.enum or [self.default])
+        return self
 
 
 class Server(ObjectExtended):

src/aiopenapi3/v31/info.py

@@ -27,13 +27,12 @@ class License(ObjectExtended):
     url: str | None = Field(default=None)
 
     @model_validator(mode="after")
-    @classmethod
-    def validate_License(cls, l: "License"):
+    def validate_License(self):
         """
         A URL to the license used for the API. This MUST be in the form of a URL. The url field is mutually exclusive of the identifier field.
         """
-        assert not all([getattr(l, i, None) is not None for i in ["identifier", "url"]])
-        return l
+        assert not all([getattr(self, i, None) is not None for i in ["identifier", "url"]])
+        return self
 
 
 class Info(ObjectExtended):

src/aiopenapi3/v31/paths.py

@@ -38,14 +38,14 @@ class Link(ObjectExtended):
     server: Server | None = Field(default=None)
 
     @model_validator(mode="after")
-    def validate_Link_operation(cls, l: '__types["Link"]'):  # noqa: F821
-        assert not (l.operationId is not None and l.operationRef is not None), (
+    def validate_Link_operation(self):  # type: ignore[name-defined]
+        assert not (self.operationId is not None and self.operationRef is not None), (
             "operationId and operationRef are mutually exclusive, only one of them is allowed"
         )
-        assert not (l.operationId == l.operationRef is None), (
+        assert not (self.operationId == self.operationRef is None), (
             "operationId and operationRef are mutually exclusive, one of them must be specified"
         )
-        return l
+        return self
 
 
 class Response(ObjectExtended):

src/aiopenapi3/v31/root.py

@@ -35,10 +35,9 @@ class Root(ObjectExtended, RootBase):
     externalDocs: dict[Any, Any] = Field(default_factory=dict)
 
     @model_validator(mode="after")
-    @classmethod
-    def validate_Root(cls, r: "Root") -> "Self":  # noqa: F821
-        assert r.paths or r.components or r.webhooks
-        return r
+    def validate_Root(self) -> "Self":  # noqa: F821
+        assert self.paths or self.components or self.webhooks
+        return self
 
     def _resolve_references(self, api):
         RootBase.resolve(api, self, self, PathItem, Reference)

src/aiopenapi3/v31/schemas.py

@@ -172,13 +172,12 @@ def is_boolean_schema(cls, data: Any) -> Any:
             return {"not": {}}
 
     @model_validator(mode="after")
-    @classmethod
-    def validate_Schema_number_type(cls, s: "Schema"):
-        if s.type == "integer":
+    def validate_Schema_number_type(self):
+        if self.type == "integer":
             for i in ["minimum", "maximum"]:
-                if (v := getattr(s, i, None)) is not None and not isinstance(v, int):
-                    setattr(s, i, int(v))
-        return s
+                if (v := getattr(self, i, None)) is not None and not isinstance(v, int):
+                    setattr(self, i, int(v))
+        return self
 
     def __getstate__(self):
         return SchemaBase.__getstate__(self)

src/aiopenapi3/v31/servers.py

@@ -17,11 +17,11 @@ class ServerVariable(ObjectExtended):
     description: str | None = Field(default=None)
 
     @model_validator(mode="after")
-    def validate_ServerVariable(cls, s: "ServerVariable"):
-        assert isinstance(s.enum, (list, None.__class__))
+    def validate_ServerVariable(self):
+        assert isinstance(self.enum, (list, None.__class__))
         # default value must be in enum
-        assert s.default is None or s.default in (s.enum or [s.default])
-        return s
+        assert self.default is None or self.default in (self.enum or [self.default])
+        return self
 
 
 class Server(ObjectExtended):

tests/apiv2_test.py

@@ -125,18 +125,18 @@ async def test_model(server, client):
 
 def randomPet(client, name=None, cat=False):
     if name:
-        Pet = client.components.schemas["Pet-Input"].get_type()
+        Pet = client.components.schemas["Pet"].get_type()
         if not cat:
-            Dog = typing.get_args(typing.get_args(Pet.model_fields["root"].annotation)[0])[1]
+            Dog = typing.get_args(Pet.model_fields["root"].annotation)[1]
             dog = Dog(
                 name=name,
                 age=datetime.timedelta(seconds=random.randint(1, 2**32)),
                 tags=[],
             )
             pet = Pet(dog)
         else:
-            Cat = typing.get_args(typing.get_args(Pet.model_fields["root"].annotation)[0])[0]
-            WhiteCat = typing.get_args(typing.get_args(Cat.model_fields["root"].annotation)[0])[1]
+            Cat = typing.get_args(Pet.model_fields["root"].annotation)[0]
+            WhiteCat = typing.get_args(Cat.model_fields["root"].annotation)[1]
             wc = WhiteCat(pet_type="cat", color="white", name="whitey", white_name="white")
             cat = Cat(wc)
             pet = Pet(cat)
@@ -236,8 +236,8 @@ async def test_deletePet(server, client):
 
 @pytest.mark.asyncio(loop_scope="session")
 async def test_patchPet(server, client):
-    Pet = client.components.schemas["Pet-Input"].get_type()
-    Dog = typing.get_args(typing.get_args(Pet.model_fields["root"].annotation)[0])[1]
+    Pet = client.components.schemas["Pet"].get_type()
+    Dog = typing.get_args(Pet.model_fields["root"].annotation)[1]
     pets = [
         Pet(
             Dog.model_construct(

tests/fixtures/schema-type-validators.yaml

@@ -9,6 +9,7 @@ components:
       type: integer
       maximum: 10
       minimum: 10
+      default: 10
     Number:
       type: number
       maximum: 10
@@ -22,3 +23,4 @@ components:
       minLength: 5
       maximum: 10
       minimum: 10
+      default: 10

tests/schema_test.py

@@ -3,7 +3,6 @@
 import uuid
 from datetime import datetime
 
-from pydantic.fields import FieldInfo
 
 from pathlib import Path
 
@@ -121,12 +120,10 @@ def test_schema_regex_engine(with_schema_regex_engine):
     import pydantic_core._pydantic_core
 
     with pytest.raises(pydantic_core._pydantic_core.SchemaError, match="error: unclosed character class$"):
-        annotations = typing.get_args(Root.model_fields["root"].annotation)
-        assert len(annotations) == 2 and annotations[0] is str and isinstance(annotations[1], FieldInfo), annotations
-        metadata = annotations[1].metadata
-        assert len(metadata) == 1, metadata
-        pattern = metadata[0].pattern
-        assert isinstance(pattern, str), pattern
+        t = Root.model_fields["root"]
+        assert t.annotation is str
+        assert isinstance(t.metadata[0].pattern, str)
+        pattern = t.metadata[0].pattern
         from typing import Annotated
 
         C = Annotated[str, pydantic.Field(pattern=pattern)]
@@ -655,6 +652,8 @@ def test_schema_type_validators(with_schema_type_validators):
     v = t.model_validate("10")
     with pytest.raises(ValidationError):
         v = t.model_validate("9")
+    v = t()
+    assert v.root == 10
 
     t = (m := api.components.schemas["Number"]).get_type()
     v = t.model_validate("10.")
@@ -680,6 +679,9 @@ def test_schema_type_validators(with_schema_type_validators):
     with pytest.raises(ValidationError):
         v = t.model_validate("invalid")
 
+    v = t()
+    assert v.root == 10
+
 
 def test_schema_allof_string(with_schema_allof_string):
     api = OpenAPI("/", with_schema_allof_string)