chore: switch type hints to use | rather than Union

Signed-off-by: Tudor Plugaru <[email protected]>

Author: PlugaruT

src/cloudevents/core/base.py

@@ -12,9 +12,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-
 from datetime import datetime
-from typing import Any, Optional, Protocol, Union
+from typing import Any, Protocol
 
 
 class BaseCloudEvent(Protocol):
@@ -29,7 +28,7 @@ class BaseCloudEvent(Protocol):
     def __init__(
         self,
         attributes: dict[str, Any],
-        data: Optional[Union[dict[str, Any], str, bytes]] = None,
+        data: dict[str, Any] | str | bytes | None = None,
     ) -> None:
         """
         Create a new CloudEvent instance.
@@ -74,31 +73,31 @@ def get_specversion(self) -> str:
         """
         ...
 
-    def get_datacontenttype(self) -> Optional[str]:
+    def get_datacontenttype(self) -> str | None:
         """
         Retrieve the datacontenttype of the event.
 
         :return: The datacontenttype of the event.
         """
         ...
 
-    def get_dataschema(self) -> Optional[str]:
+    def get_dataschema(self) -> str | None:
         """
         Retrieve the dataschema of the event.
 
         :return: The dataschema of the event.
         """
         ...
 
-    def get_subject(self) -> Optional[str]:
+    def get_subject(self) -> str | None:
         """
         Retrieve the subject of the event.
 
         :return: The subject of the event.
         """
         ...
 
-    def get_time(self) -> Optional[datetime]:
+    def get_time(self) -> datetime | None:
         """
         Retrieve the time of the event.
 
@@ -115,7 +114,7 @@ def get_extension(self, extension_name: str) -> Any:
         """
         ...
 
-    def get_data(self) -> Optional[Union[dict[str, Any], str, bytes]]:
+    def get_data(self) -> dict[str, Any] | str | bytes | None:
         """
         Retrieve data of the event.
 

src/cloudevents/core/bindings/http.py

@@ -14,7 +14,7 @@
 
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Callable, Final, Optional, Union
+from typing import Any, Callable, Final
 from urllib.parse import quote, unquote
 
 from dateutil.parser import isoparse
@@ -140,7 +140,7 @@ def from_binary(
     message: HTTPMessage,
     event_format: Format,
     event_factory: Callable[
-        [dict[str, Any], Optional[Union[dict[str, Any], str, bytes]]], BaseCloudEvent
+        [dict[str, Any], dict[str, Any] | str | bytes | None], BaseCloudEvent
     ],
 ) -> BaseCloudEvent:
     """
@@ -221,7 +221,7 @@ def from_structured(
     message: HTTPMessage,
     event_format: Format,
     event_factory: Callable[
-        [dict[str, Any], Optional[Union[dict[str, Any], str, bytes]]], BaseCloudEvent
+        [dict[str, Any], dict[str, Any] | str | bytes | None], BaseCloudEvent
     ],
 ) -> BaseCloudEvent:
     """
@@ -252,7 +252,7 @@ def from_http(
     message: HTTPMessage,
     event_format: Format,
     event_factory: Callable[
-        [dict[str, Any], Optional[Union[dict[str, Any], str, bytes]]], BaseCloudEvent
+        [dict[str, Any], dict[str, Any] | str | bytes | None], BaseCloudEvent
     ],
 ) -> BaseCloudEvent:
     """

src/cloudevents/core/formats/base.py

@@ -12,8 +12,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-
-from typing import Any, Callable, Optional, Protocol, Union
+from typing import Any, Callable, Protocol
 
 from cloudevents.core.base import BaseCloudEvent
 
@@ -31,10 +30,10 @@ class Format(Protocol):
     def read(
         self,
         event_factory: Callable[
-            [dict[str, Any], Optional[Union[dict[str, Any], str, bytes]]],
+            [dict[str, Any], dict[str, Any] | str | bytes | None],
             BaseCloudEvent,
         ],
-        data: Union[str, bytes],
+        data: str | bytes,
     ) -> BaseCloudEvent:
         """
         Deserialize a CloudEvent from its wire format representation.
@@ -60,8 +59,8 @@ def write(self, event: BaseCloudEvent) -> bytes:
 
     def write_data(
         self,
-        data: Optional[Union[dict[str, Any], str, bytes]],
-        datacontenttype: Optional[str],
+        data: dict[str, Any] | str | bytes | None,
+        datacontenttype: str | None,
     ) -> bytes:
         """
         Serialize just the data payload for protocol bindings (e.g., HTTP binary mode).
@@ -73,8 +72,8 @@ def write_data(
         ...
 
     def read_data(
-        self, body: bytes, datacontenttype: Optional[str]
-    ) -> Optional[Union[dict[str, Any], str, bytes]]:
+        self, body: bytes, datacontenttype: str | None
+    ) -> dict[str, Any] | str | bytes | None:
         """
         Deserialize data payload from protocol bindings (e.g., HTTP binary mode).
 

src/cloudevents/core/formats/json.py

@@ -12,12 +12,11 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-
 import base64
 import re
 from datetime import datetime
 from json import JSONEncoder, dumps, loads
-from typing import Any, Callable, Final, Optional, Pattern, Union
+from typing import Any, Callable, Final, Pattern
 
 from dateutil.parser import isoparse
 
@@ -51,10 +50,10 @@ class JSONFormat(Format):
     def read(
         self,
         event_factory: Callable[
-            [dict[str, Any], Optional[Union[dict[str, Any], str, bytes]]],
+            [dict[str, Any], dict[str, Any] | str | bytes | None],
             BaseCloudEvent,
         ],
-        data: Union[str, bytes],
+        data: str | bytes,
     ) -> BaseCloudEvent:
         """
         Read a CloudEvent from a JSON formatted byte string.
@@ -74,7 +73,7 @@ def read(
         if "time" in event_attributes:
             event_attributes["time"] = isoparse(event_attributes["time"])
 
-        event_data: Union[dict[str, Any], str, bytes, None] = event_attributes.pop(
+        event_data: dict[str, Any] | str | bytes | None = event_attributes.pop(
             "data", None
         )
         if event_data is None:
@@ -108,8 +107,8 @@ def write(self, event: BaseCloudEvent) -> bytes:
 
     def write_data(
         self,
-        data: Optional[Union[dict[str, Any], str, bytes]],
-        datacontenttype: Optional[str],
+        data: dict[str, Any] | str | bytes | None,
+        datacontenttype: str | None,
     ) -> bytes:
         """
         Serialize just the data payload for HTTP binary mode.
@@ -143,8 +142,8 @@ def write_data(
         return str(data).encode("utf-8")
 
     def read_data(
-        self, body: bytes, datacontenttype: Optional[str]
-    ) -> Optional[Union[dict[str, Any], str, bytes]]:
+        self, body: bytes, datacontenttype: str | None
+    ) -> dict[str, Any] | str | bytes | None:
         """
         Deserialize data payload from HTTP binary mode body.
 

src/cloudevents/core/v1/event.py

@@ -15,7 +15,7 @@
 import re
 from collections import defaultdict
 from datetime import datetime
-from typing import Any, Final, Optional, Union
+from typing import Any, Final
 
 from cloudevents.core.base import BaseCloudEvent
 from cloudevents.core.v1.exceptions import (
@@ -40,11 +40,11 @@ class CloudEvent(BaseCloudEvent):
     def __init__(
         self,
         attributes: dict[str, Any],
-        data: Optional[Union[dict[str, Any], str, bytes]] = None,
+        data: dict[str, Any] | str | bytes | None = None,
     ) -> None:
         self._validate_attribute(attributes=attributes)
         self._attributes: dict[str, Any] = attributes
-        self._data: Optional[Union[dict[str, Any], str, bytes]] = data
+        self._data: dict[str, Any] | str | bytes | None = data
 
     @staticmethod
     def _validate_attribute(attributes: dict[str, Any]) -> None:
@@ -242,22 +242,22 @@ def get_type(self) -> str:
     def get_specversion(self) -> str:
         return self._attributes["specversion"]  # type: ignore
 
-    def get_datacontenttype(self) -> Optional[str]:
+    def get_datacontenttype(self) -> str | None:
         return self._attributes.get("datacontenttype")
 
-    def get_dataschema(self) -> Optional[str]:
+    def get_dataschema(self) -> str | None:
         return self._attributes.get("dataschema")
 
-    def get_subject(self) -> Optional[str]:
+    def get_subject(self) -> str | None:
         return self._attributes.get("subject")
 
-    def get_time(self) -> Optional[datetime]:
+    def get_time(self) -> datetime | None:
         return self._attributes.get("time")
 
     def get_extension(self, extension_name: str) -> Any:
         return self._attributes.get(extension_name)
 
-    def get_data(self) -> Optional[Union[dict[str, Any], str, bytes]]:
+    def get_data(self) -> dict[str, Any] | str | bytes | None:
         return self._data
 
     def get_attributes(self) -> dict[str, Any]:

tests/test_core/test_bindings/test_http.py

@@ -13,7 +13,7 @@
 #    under the License.
 
 from datetime import datetime, timezone
-from typing import Any, Optional, Union
+from typing import Any
 
 import pytest
 
@@ -41,8 +41,8 @@ def minimal_attributes() -> dict[str, str]:
 
 
 def create_event(
-    extra_attrs: Optional[dict[str, Any]] = None,
-    data: Optional[Union[dict[str, Any], str, bytes]] = None,
+    extra_attrs: dict[str, Any] | None = None,
+    data: dict[str, Any] | str | bytes | None = None,
 ) -> CloudEvent:
     """Helper to create CloudEvent with valid required attributes"""
     attrs: dict[str, Any] = {
@@ -908,7 +908,7 @@ def test_custom_event_factory() -> None:
     """Test using custom event factory function"""
 
     def custom_factory(
-        attributes: dict[str, Any], data: Optional[Union[dict[str, Any], str, bytes]]
+        attributes: dict[str, Any], data: dict[str, Any] | str | bytes | None
     ) -> CloudEvent:
         # Custom factory that adds a prefix to the type
         attributes["type"] = f"custom.{attributes.get('type', 'unknown')}"