cloudevents
/
sdk-python
mirror of https://github.com/cloudevents/sdk-python.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Pydantic v2 (#235) 

* Fixes examples when using Pydantic V2

Signed-off-by: Fabio Batista <fabio@atelie.dev.br>

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

Signed-off-by: Fabio Batista <fabio@atelie.dev.br>

---------

Signed-off-by: Fabio Batista <fabio@atelie.dev.br>
Co-authored-by: Yurii Serhiichuk <xSAVIKx@users.noreply.github.com>
This commit is contained in:
Fábio D. Batista 2024-05-26 15:51:36 -03:00 committed by GitHub
parent eedc61e9b0
commit 11520e35e1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 31 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.pre-commit-config.yaml
View File

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

33
cloudevents/pydantic/__init__.py
View File

@ -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(

18
cloudevents/pydantic/v2/event.py
View File

@ -51,53 +51,53 @@ class CloudEvent(abstract.CloudEvent, BaseModel): # type: ignore
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,
)