Pydantic v2 (#235)

* Fixes examples when using Pydantic V2

Signed-off-by: Fabio Batista <[email protected]>

* When type checking, uses the latest (V2) version of Pydantic

Signed-off-by: Fabio Batista <[email protected]>

---------

Signed-off-by: Fabio Batista <[email protected]>
Co-authored-by: Yurii Serhiichuk <[email protected]>

Author: fabiob

.pre-commit-config.yaml

@@ -24,4 +24,4 @@ repos:
         types: [ python ]
         args: [ ]
         additional_dependencies:
-          - "pydantic"
+          - "pydantic~=2.7"

cloudevents/pydantic/__init__.py

@@ -12,22 +12,31 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from typing import TYPE_CHECKING
+
 from cloudevents.exceptions import PydanticFeatureNotInstalled
 
 try:
-    from pydantic import VERSION as PYDANTIC_VERSION
-
-    pydantic_major_version = PYDANTIC_VERSION.split(".")[0]
-    if pydantic_major_version == "1":
-        from cloudevents.pydantic.v1 import CloudEvent, from_dict, from_http, from_json
-
+    if TYPE_CHECKING:
+        from cloudevents.pydantic.v2 import CloudEvent, from_dict, from_http, from_json
     else:
-        from cloudevents.pydantic.v2 import (  # type: ignore
-            CloudEvent,
-            from_dict,
-            from_http,
-            from_json,
-        )
+        from pydantic import VERSION as PYDANTIC_VERSION
+
+        pydantic_major_version = PYDANTIC_VERSION.split(".")[0]
+        if pydantic_major_version == "1":
+            from cloudevents.pydantic.v1 import (
+                CloudEvent,
+                from_dict,
+                from_http,
+                from_json,
+            )
+        else:
+            from cloudevents.pydantic.v2 import (
+                CloudEvent,
+                from_dict,
+                from_http,
+                from_json,
+            )
 
 except ImportError:  # pragma: no cover # hard to test
     raise PydanticFeatureNotInstalled(

cloudevents/pydantic/v2/event.py

@@ -51,53 +51,53 @@ def create(
     data: typing.Optional[typing.Any] = Field(
         title=FIELD_DESCRIPTIONS["data"].get("title"),
         description=FIELD_DESCRIPTIONS["data"].get("description"),
-        example=FIELD_DESCRIPTIONS["data"].get("example"),
+        examples=[FIELD_DESCRIPTIONS["data"].get("example")],
         default=None,
     )
     source: str = Field(
         title=FIELD_DESCRIPTIONS["source"].get("title"),
         description=FIELD_DESCRIPTIONS["source"].get("description"),
-        example=FIELD_DESCRIPTIONS["source"].get("example"),
+        examples=[FIELD_DESCRIPTIONS["source"].get("example")],
     )
     id: str = Field(
         title=FIELD_DESCRIPTIONS["id"].get("title"),
         description=FIELD_DESCRIPTIONS["id"].get("description"),
-        example=FIELD_DESCRIPTIONS["id"].get("example"),
+        examples=[FIELD_DESCRIPTIONS["id"].get("example")],
         default_factory=attribute.default_id_selection_algorithm,
     )
     type: str = Field(
         title=FIELD_DESCRIPTIONS["type"].get("title"),
         description=FIELD_DESCRIPTIONS["type"].get("description"),
-        example=FIELD_DESCRIPTIONS["type"].get("example"),
+        examples=[FIELD_DESCRIPTIONS["type"].get("example")],
     )
     specversion: attribute.SpecVersion = Field(
         title=FIELD_DESCRIPTIONS["specversion"].get("title"),
         description=FIELD_DESCRIPTIONS["specversion"].get("description"),
-        example=FIELD_DESCRIPTIONS["specversion"].get("example"),
+        examples=[FIELD_DESCRIPTIONS["specversion"].get("example")],
         default=attribute.DEFAULT_SPECVERSION,
     )
     time: typing.Optional[datetime.datetime] = Field(
         title=FIELD_DESCRIPTIONS["time"].get("title"),
         description=FIELD_DESCRIPTIONS["time"].get("description"),
-        example=FIELD_DESCRIPTIONS["time"].get("example"),
+        examples=[FIELD_DESCRIPTIONS["time"].get("example")],
         default_factory=attribute.default_time_selection_algorithm,
     )
     subject: typing.Optional[str] = Field(
         title=FIELD_DESCRIPTIONS["subject"].get("title"),
         description=FIELD_DESCRIPTIONS["subject"].get("description"),
-        example=FIELD_DESCRIPTIONS["subject"].get("example"),
+        examples=[FIELD_DESCRIPTIONS["subject"].get("example")],
         default=None,
     )
     datacontenttype: typing.Optional[str] = Field(
         title=FIELD_DESCRIPTIONS["datacontenttype"].get("title"),
         description=FIELD_DESCRIPTIONS["datacontenttype"].get("description"),
-        example=FIELD_DESCRIPTIONS["datacontenttype"].get("example"),
+        examples=[FIELD_DESCRIPTIONS["datacontenttype"].get("example")],
         default=None,
     )
     dataschema: typing.Optional[str] = Field(
         title=FIELD_DESCRIPTIONS["dataschema"].get("title"),
         description=FIELD_DESCRIPTIONS["dataschema"].get("description"),
-        example=FIELD_DESCRIPTIONS["dataschema"].get("example"),
+        examples=[FIELD_DESCRIPTIONS["dataschema"].get("example")],
         default=None,
     )