chore: switch from using Dict to dict for type hints

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

Author: PlugaruT

src/cloudevents/core/base.py

@@ -14,7 +14,7 @@
 
 
 from datetime import datetime
-from typing import Any, Dict, Optional, Protocol, Union
+from typing import Any, Optional, Protocol, Union
 
 
 class BaseCloudEvent(Protocol):
@@ -28,8 +28,8 @@ class BaseCloudEvent(Protocol):
 
     def __init__(
         self,
-        attributes: Dict[str, Any],
-        data: Optional[Union[Dict[str, Any], str, bytes]] = None,
+        attributes: dict[str, Any],
+        data: Optional[Union[dict[str, Any], str, bytes]] = None,
     ) -> None:
         """
         Create a new CloudEvent instance.
@@ -115,7 +115,7 @@ def get_extension(self, extension_name: str) -> Any:
         """
         ...
 
-    def get_data(self) -> Optional[Union[Dict[str, Any], str, bytes]]:
+    def get_data(self) -> Optional[Union[dict[str, Any], str, bytes]]:
         """
         Retrieve data of the event.
 

src/cloudevents/core/bindings/http.py

@@ -14,16 +14,16 @@
 
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Callable, Dict, Optional, Union
+from typing import Any, Callable, Final, Optional, Union
 from urllib.parse import quote, unquote
 
 from dateutil.parser import isoparse
 
 from cloudevents.core.base import BaseCloudEvent
 from cloudevents.core.formats.base import Format
 
-CE_PREFIX: str = "ce-"
-CONTENT_TYPE_HEADER: str = "content-type"
+CE_PREFIX: Final[str] = "ce-"
+CONTENT_TYPE_HEADER: Final[str] = "content-type"
 
 
 @dataclass(frozen=True)
@@ -40,11 +40,11 @@ class HTTPMessage:
         body: HTTP body as bytes
     """
 
-    headers: Dict[str, str]
+    headers: dict[str, str]
     body: bytes
 
 
-def _normalize_headers(headers: Dict[str, str]) -> Dict[str, str]:
+def _normalize_headers(headers: dict[str, str]) -> dict[str, str]:
     """
     Normalize HTTP headers by converting all keys to lowercase.
 
@@ -116,7 +116,7 @@ def to_binary(event: BaseCloudEvent, event_format: Format) -> HTTPMessage:
     :param event_format: Format implementation for data serialization
     :return: HTTPMessage with ce-prefixed headers and event data as body
     """
-    headers: Dict[str, str] = {}
+    headers: dict[str, str] = {}
     attributes = event.get_attributes()
 
     for attr_name, attr_value in attributes.items():
@@ -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], Optional[Union[dict[str, Any], str, bytes]]], BaseCloudEvent
     ],
 ) -> BaseCloudEvent:
     """
@@ -168,7 +168,7 @@ def from_binary(
     """
     normalized_headers = _normalize_headers(message.headers)
 
-    attributes: Dict[str, Any] = {}
+    attributes: dict[str, Any] = {}
 
     for header_name, header_value in normalized_headers.items():
         if header_name.startswith(CE_PREFIX):
@@ -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], Optional[Union[dict[str, Any], str, bytes]]], 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], Optional[Union[dict[str, Any], str, bytes]]], BaseCloudEvent
     ],
 ) -> BaseCloudEvent:
     """

src/cloudevents/core/formats/base.py

@@ -13,7 +13,7 @@
 #    under the License.
 
 
-from typing import Any, Callable, Dict, Optional, Protocol, Union
+from typing import Any, Callable, Optional, Protocol, Union
 
 from cloudevents.core.base import BaseCloudEvent
 
@@ -31,7 +31,7 @@ class Format(Protocol):
     def read(
         self,
         event_factory: Callable[
-            [Dict[str, Any], Optional[Union[Dict[str, Any], str, bytes]]],
+            [dict[str, Any], Optional[Union[dict[str, Any], str, bytes]]],
             BaseCloudEvent,
         ],
         data: Union[str, bytes],
@@ -60,7 +60,7 @@ def write(self, event: BaseCloudEvent) -> bytes:
 
     def write_data(
         self,
-        data: Optional[Union[Dict[str, Any], str, bytes]],
+        data: Optional[Union[dict[str, Any], str, bytes]],
         datacontenttype: Optional[str],
     ) -> bytes:
         """
@@ -74,7 +74,7 @@ def write_data(
 
     def read_data(
         self, body: bytes, datacontenttype: Optional[str]
-    ) -> Optional[Union[Dict[str, Any], str, bytes]]:
+    ) -> Optional[Union[dict[str, Any], str, bytes]]:
         """
         Deserialize data payload from protocol bindings (e.g., HTTP binary mode).
 

src/cloudevents/core/formats/json.py

@@ -17,7 +17,7 @@
 import re
 from datetime import datetime
 from json import JSONEncoder, dumps, loads
-from typing import Any, Callable, Dict, Final, Optional, Pattern, Union
+from typing import Any, Callable, Final, Optional, Pattern, Union
 
 from dateutil.parser import isoparse
 
@@ -51,7 +51,7 @@ class JSONFormat(Format):
     def read(
         self,
         event_factory: Callable[
-            [Dict[str, Any], Optional[Union[Dict[str, Any], str, bytes]]],
+            [dict[str, Any], Optional[Union[dict[str, Any], str, bytes]]],
             BaseCloudEvent,
         ],
         data: Union[str, bytes],
@@ -74,7 +74,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: Union[dict[str, Any], str, bytes, None] = event_attributes.pop(
             "data", None
         )
         if event_data is None:
@@ -108,7 +108,7 @@ def write(self, event: BaseCloudEvent) -> bytes:
 
     def write_data(
         self,
-        data: Optional[Union[Dict[str, Any], str, bytes]],
+        data: Optional[Union[dict[str, Any], str, bytes]],
         datacontenttype: Optional[str],
     ) -> bytes:
         """
@@ -144,7 +144,7 @@ def write_data(
 
     def read_data(
         self, body: bytes, datacontenttype: Optional[str]
-    ) -> Optional[Union[Dict[str, Any], str, bytes]]:
+    ) -> Optional[Union[dict[str, Any], str, bytes]]:
         """
         Deserialize data payload from HTTP binary mode body.
 
@@ -164,7 +164,7 @@ def read_data(
         ):
             try:
                 decoded = body.decode("utf-8")
-                parsed: Dict[str, Any] = loads(decoded)
+                parsed: dict[str, Any] = loads(decoded)
                 return parsed
             except (ValueError, UnicodeDecodeError):
                 # If JSON parsing fails, fall through to other handling

src/cloudevents/core/v1/event.py

@@ -15,7 +15,7 @@
 import re
 from collections import defaultdict
 from datetime import datetime
-from typing import Any, Dict, Final, List, Optional, Union
+from typing import Any, Final, Optional, Union
 
 from cloudevents.core.base import BaseCloudEvent
 from cloudevents.core.v1.exceptions import (
@@ -39,15 +39,15 @@
 class CloudEvent(BaseCloudEvent):
     def __init__(
         self,
-        attributes: Dict[str, Any],
-        data: Optional[Union[Dict[str, Any], str, bytes]] = None,
+        attributes: dict[str, Any],
+        data: Optional[Union[dict[str, Any], str, bytes]] = None,
     ) -> None:
         self._validate_attribute(attributes=attributes)
-        self._attributes: Dict[str, Any] = attributes
-        self._data: Optional[Union[Dict[str, Any], str, bytes]] = data
+        self._attributes: dict[str, Any] = attributes
+        self._data: Optional[Union[dict[str, Any], str, bytes]] = data
 
     @staticmethod
-    def _validate_attribute(attributes: Dict[str, Any]) -> None:
+    def _validate_attribute(attributes: dict[str, Any]) -> None:
         """
         Validates the attributes of the CloudEvent as per the CloudEvents specification.
 
@@ -62,15 +62,15 @@ def _validate_attribute(attributes: Dict[str, Any]) -> None:
 
     @staticmethod
     def _validate_required_attributes(
-        attributes: Dict[str, Any],
-    ) -> Dict[str, List[BaseCloudEventException]]:
+        attributes: dict[str, Any],
+    ) -> dict[str, list[BaseCloudEventException]]:
         """
         Validates the types of the required attributes.
 
         :param attributes: The attributes of the CloudEvent instance.
         :return: A dictionary of validation error messages.
         """
-        errors: Dict[str, List[BaseCloudEventException]] = defaultdict(list)
+        errors: dict[str, list[BaseCloudEventException]] = defaultdict(list)
 
         if "id" not in attributes:
             errors["id"].append(MissingRequiredAttributeError(attribute_name="id"))
@@ -122,15 +122,15 @@ def _validate_required_attributes(
 
     @staticmethod
     def _validate_optional_attributes(
-        attributes: Dict[str, Any],
-    ) -> Dict[str, List[BaseCloudEventException]]:
+        attributes: dict[str, Any],
+    ) -> dict[str, list[BaseCloudEventException]]:
         """
         Validates the types and values of the optional attributes.
 
         :param attributes: The attributes of the CloudEvent instance.
         :return: A dictionary of validation error messages.
         """
-        errors: Dict[str, List[BaseCloudEventException]] = defaultdict(list)
+        errors: dict[str, list[BaseCloudEventException]] = defaultdict(list)
 
         if "time" in attributes:
             if not isinstance(attributes["time"], datetime):
@@ -192,15 +192,15 @@ def _validate_optional_attributes(
 
     @staticmethod
     def _validate_extension_attributes(
-        attributes: Dict[str, Any],
-    ) -> Dict[str, List[BaseCloudEventException]]:
+        attributes: dict[str, Any],
+    ) -> dict[str, list[BaseCloudEventException]]:
         """
         Validates the extension attributes.
 
         :param attributes: The attributes of the CloudEvent instance.
         :return: A dictionary of validation error messages.
         """
-        errors: Dict[str, List[BaseCloudEventException]] = defaultdict(list)
+        errors: dict[str, list[BaseCloudEventException]] = defaultdict(list)
         extension_attributes = [
             key
             for key in attributes.keys()
@@ -257,8 +257,8 @@ def get_time(self) -> Optional[datetime]:
     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) -> Optional[Union[dict[str, Any], str, bytes]]:
         return self._data
 
-    def get_attributes(self) -> Dict[str, Any]:
+    def get_attributes(self) -> dict[str, Any]:
         return self._attributes

tests/test_core/test_bindings/test_http.py

@@ -13,7 +13,7 @@
 #    under the License.
 
 from datetime import datetime, timezone
-from typing import Any, Callable, Dict, Optional, Union
+from typing import Any, Optional, Union
 
 import pytest
 
@@ -30,7 +30,7 @@
 
 
 @pytest.fixture
-def minimal_attributes() -> Dict[str, str]:
+def minimal_attributes() -> dict[str, str]:
     """Minimal valid CloudEvent attributes"""
     return {
         "type": "com.example.test",
@@ -41,11 +41,11 @@ 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: Optional[dict[str, Any]] = None,
+    data: Optional[Union[dict[str, Any], str, bytes]] = None,
 ) -> CloudEvent:
     """Helper to create CloudEvent with valid required attributes"""
-    attrs: Dict[str, Any] = {
+    attrs: dict[str, Any] = {
         "type": "com.example.test",
         "source": "/test",
         "id": "test-id-123",
@@ -68,10 +68,10 @@ def test_http_message_immutable() -> None:
     message = HTTPMessage(headers={"test": "value"}, body=b"data")
 
     with pytest.raises(Exception):  # FrozenInstanceError
-        message.headers = {"new": "dict"}  # type: ignore[misc]
+        message.headers = {"new": "dict"}
 
     with pytest.raises(Exception):  # FrozenInstanceError
-        message.body = b"new data"  # type: ignore[misc]
+        message.body = b"new data"
 
 
 def test_http_message_with_empty_headers() -> None:
@@ -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: Optional[Union[dict[str, Any], str, bytes]]
     ) -> CloudEvent:
         # Custom factory that adds a prefix to the type
         attributes["type"] = f"custom.{attributes.get('type', 'unknown')}"