Use TypeAdapters for single event validation

Signed-off-by: Federico Busetti <729029+febus982@users.noreply.github.com>

cloudevents_pydantic/bindings/http.py

@@ -26,16 +26,16 @@
     List,
     NamedTuple,
     Type,
+    TypeVar,
     cast,
 )
 
 from pydantic import TypeAdapter
-from typing_extensions import TypeVar
 
 from cloudevents_pydantic.events import CloudEvent
 from cloudevents_pydantic.formats import json
 
-_T = TypeVar("_T", bound=CloudEvent, default=CloudEvent)
+_T = TypeVar("_T", bound=CloudEvent)
 
 
 class HTTPComponents(NamedTuple):
@@ -44,12 +44,12 @@ class HTTPComponents(NamedTuple):
 
 
 class HTTPHandler(Generic[_T]):
-    event_class: Type[_T]
+    event_adapter: TypeAdapter[_T]
     batch_adapter: TypeAdapter[List[_T]]
 
     def __init__(self, event_class: Type[_T] = cast(Type[_T], CloudEvent)) -> None:
         super().__init__()
-        self.event_class = event_class
+        self.event_adapter = TypeAdapter(event_class)
         self.batch_adapter = TypeAdapter(List[event_class])  # type: ignore[valid-type]
 
     def to_json(self, event: _T) -> HTTPComponents:
@@ -90,7 +90,7 @@ def from_json(
         :return: The deserialized event
         :rtype: CloudEvent
         """
-        return json.from_json(body, self.event_class)
+        return json.from_json(body, self.event_adapter)
 
     def from_json_batch(
         self,

cloudevents_pydantic/formats/json.py

@@ -20,14 +20,13 @@
 #  FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER         =
 #  DEALINGS IN THE SOFTWARE.                                                   =
 # ==============================================================================
-from typing import List, Type, overload
+from typing import List, TypeVar
 
 from pydantic import TypeAdapter
-from typing_extensions import TypeVar
 
 from ..events import CloudEvent
 
-_T = TypeVar("_T", bound=CloudEvent, default=CloudEvent)
+_T = TypeVar("_T", bound=CloudEvent)
 
 
 def to_json(event: CloudEvent) -> str:
@@ -42,22 +41,20 @@ def to_json(event: CloudEvent) -> str:
     return event.model_dump_json()
 
 
-@overload
-def from_json(data: str) -> CloudEvent: ...
-@overload
-def from_json(data: str, event_class: Type[_T]) -> _T: ...
-def from_json(data: str, event_class: Type[CloudEvent] = CloudEvent) -> CloudEvent:
+def from_json(
+    data: str, event_adapter: TypeAdapter[_T] = TypeAdapter(CloudEvent)
+) -> _T:
     """
     Deserializes an event from JSON format.
 
     :param data: the JSON representation of the event
     :type data: str
-    :param event_class: The event class to build
-    :type event_class: Type[CloudEvent]
+    :param event_adapter: The event class to build
+    :type event_adapter: Type[CloudEvent]
     :return: The deserialized event
     :rtype: CloudEvent
     """
-    return event_class.model_validate_json(data)
+    return event_adapter.validate_json(data)
 
 
 def to_json_batch(

tests/bindings/test_http.py

@@ -21,7 +21,7 @@
 #  DEALINGS IN THE SOFTWARE.                                                   =
 # ==============================================================================
 from typing import List
-from unittest.mock import patch
+from unittest.mock import MagicMock, call, patch
 
 import pytest
 from pydantic import Field, TypeAdapter
@@ -97,16 +97,24 @@ def from_json_batch_spy():
         yield mocked_function
 
 
-def test_initialization_defaults_to_cloudevents(type_adapter_init_mock):
-    handler = HTTPHandler()
-    assert handler.event_class is CloudEvent
-    type_adapter_init_mock.assert_called_once_with(List[CloudEvent])
+def test_initialization_defaults_to_cloudevents(type_adapter_init_mock: MagicMock):
+    HTTPHandler()
+    type_adapter_init_mock.assert_has_calls(
+        [
+            call(CloudEvent),
+            call(List[CloudEvent]),
+        ]
+    )
 
 
-def test_initialization_uses_provided_event_class(type_adapter_init_mock):
-    handler = HTTPHandler(event_class=SomeEvent)
-    assert handler.event_class is SomeEvent
-    type_adapter_init_mock.assert_called_once_with(List[SomeEvent])
+def test_initialization_uses_provided_event_class(type_adapter_init_mock: MagicMock):
+    HTTPHandler(event_class=SomeEvent)
+    type_adapter_init_mock.assert_has_calls(
+        [
+            call(SomeEvent),
+            call(List[SomeEvent]),
+        ]
+    )
 
 
 @pytest.mark.parametrize(
@@ -143,7 +151,7 @@ def test_from_json(from_json_spy):
     handler = HTTPHandler(event_class=SomeEvent)
     event = handler.from_json(valid_json)
 
-    from_json_spy.assert_called_once_with(valid_json, handler.event_class)
+    from_json_spy.assert_called_once_with(valid_json, handler.event_adapter)
     assert event == SomeEvent(**test_attributes)
     assert event != CloudEvent(**test_attributes)
     assert isinstance(event, SomeEvent)

tests/formats/test_json.py

@@ -375,7 +375,7 @@ class BinaryDataEvent(CloudEvent):
             batch_adapter=TypeAdapter(Sequence[BinaryDataEvent]),
         )[0]
     else:
-        event = from_json(json_string, BinaryDataEvent)
+        event = from_json(json_string, TypeAdapter(BinaryDataEvent))
     assert event.data == expected_value
     assert isinstance(event, BinaryDataEvent)
 
@@ -412,5 +412,5 @@ class SomeData(TypedDict):
     class BinaryNestedEvent(CloudEvent):
         data: SomeData
 
-    event: BinaryNestedEvent = from_json(json_input, BinaryNestedEvent)
+    event: BinaryNestedEvent = from_json(json_input, TypeAdapter(BinaryNestedEvent))
     assert event.data["data"] == b"\x02\x03\x05\x07"