Implement read method

Signed-off-by: Tudor Plugaru <plugaru.tudor@protonmail.com>

Author: PlugaruT

pyproject.toml

@@ -32,6 +32,7 @@ keywords = [
 ]
 dependencies = [
     "ruff>=0.6.8",
+    "python-dateutil>=2.8.2",
 ]
 
 [project.urls]
@@ -53,6 +54,7 @@ dev-dependencies = [
     "flake8-print>=5.0.0",
     "pre-commit>=3.8.0",
     "pytest-cov>=5.0.0",
+    "types-python-dateutil>=2.9.0.20241003",
 ]
 
 [tool.uv.pip]

src/cloudevents/core/base.py

@@ -18,6 +18,10 @@
 
 
 class BaseCloudEvent(Protocol):
+    def __init__(
+        self, attributes: dict[str, Any], data: Optional[Union[dict, str, bytes]] = None
+    ) -> None: ...
+
     def get_id(self) -> str: ...
 
     def get_source(self) -> str: ...

src/cloudevents/core/formats/base.py

@@ -21,4 +21,4 @@
 class Format(Protocol):
     def read(self, data: Union[str, bytes]) -> BaseCloudEvent: ...
 
-    def write(self, event: BaseCloudEvent) -> str: ...
+    def write(self, event: BaseCloudEvent) -> bytes: ...

src/cloudevents/core/formats/json.py

@@ -16,12 +16,16 @@
 import base64
 import re
 from datetime import datetime
-from json import JSONEncoder, dumps
-from typing import Any, Final, Pattern, Union
+from json import JSONEncoder, dumps, loads
+from typing import Any, Final, Pattern, Type, TypeVar, Union
+
+from dateutil.parser import isoparse
 
 from cloudevents.core.base import BaseCloudEvent
 from cloudevents.core.formats.base import Format
 
+T = TypeVar("T", bound=BaseCloudEvent)
+
 
 class _JSONEncoderWithDatetime(JSONEncoder):
     """
@@ -46,18 +50,41 @@ class JSONFormat(Format):
         r"^(application|text)\\/([a-zA-Z]+\\+)?json(;.*)*$"
     )
 
-    def read(self, data: Union[str, bytes]) -> BaseCloudEvent:
-        pass
+    def read(self, event_klass: Type[T], data: Union[str, bytes]) -> T:
+        """
+        Read a CloudEvent from a JSON formatted byte string.
+
+        :param data: The JSON formatted byte array.
+        :return: The CloudEvent instance.
+        """
+        if isinstance(data, bytes):
+            decoded_data: str = data.decode("utf-8")
+        else:
+            decoded_data = data
+
+        event_attributes = loads(decoded_data)
+
+        if "time" in event_attributes:
+            event_attributes["time"] = isoparse(event_attributes["time"])
+
+        event_data: Union[str, bytes] = event_attributes.get("data")
+        if event_data is None:
+            event_data_base64 = event_attributes.get("data_base64")
+            if event_data_base64 is not None:
+                event_data = base64.b64decode(event_data_base64)
+
+        # disable mypy due to https://github.com/python/mypy/issues/9003
+        return event_klass(event_attributes, event_data)  # type: ignore
 
-    def write(self, event: BaseCloudEvent) -> bytes:
+    def write(self, event: T) -> bytes:
         """
         Write a CloudEvent to a JSON formatted byte string.
 
         :param event: The CloudEvent to write.
         :return: The CloudEvent as a JSON formatted byte array.
         """
         event_data = event.get_data()
-        event_dict: dict[str, Any] = {**event.get_attributes()}
+        event_dict: dict[str, Any] = dict(event.get_attributes())
 
         if event_data is not None:
             if isinstance(event_data, (bytes, bytearray)):

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
+from typing import Any, Final, Optional, Union
 
 from cloudevents.core.base import BaseCloudEvent
 from cloudevents.core.v1.exceptions import (
@@ -45,7 +45,9 @@ class CloudEvent(BaseCloudEvent):
     obliged to follow this contract.
     """
 
-    def __init__(self, attributes: dict[str, Any], data: Optional[dict] = None) -> None:
+    def __init__(
+        self, attributes: dict[str, Any], data: Optional[Union[dict, str, bytes]] = None
+    ) -> None:
         """
         Create a new CloudEvent instance.
 
@@ -57,7 +59,7 @@ def __init__(self, attributes: dict[str, Any], data: Optional[dict] = None) -> N
         """
         self._validate_attribute(attributes=attributes)
         self._attributes: dict[str, Any] = attributes
-        self._data: Optional[dict] = data
+        self._data: Optional[Union[dict, str, bytes]] = data
 
     @staticmethod
     def _validate_attribute(attributes: dict[str, Any]) -> None:
@@ -316,7 +318,7 @@ def get_extension(self, extension_name: str) -> Any:
         """
         return self._attributes.get(extension_name)
 
-    def get_data(self) -> Optional[dict]:
+    def get_data(self) -> Optional[Union[dict, str, bytes]]:
         """
         Retrieve data of the event.
 

tests/test_core/test_format/test_json.py

@@ -155,3 +155,23 @@ def test_write_cloud_event_to_json_with_no_content_type_set_and_data_as_json() -
             "utf-8"
         )
     )
+
+
+def test_read_cloud_event_from_json_with_attributes_only() -> None:
+    data = '{"id": "123", "source": "source", "type": "type", "specversion": "1.0", "time": "2023-10-25T17:09:19.736166Z", "datacontenttype": "application/json", "dataschema": "http://example.com/schema", "subject": "test_subject"}'.encode(
+        "utf-8"
+    )
+    formatter = JSONFormat()
+    result = formatter.read(CloudEvent, data)
+
+    assert result.get_id() == "123"
+    assert result.get_source() == "source"
+    assert result.get_type() == "type"
+    assert result.get_specversion() == "1.0"
+    assert result.get_time() == datetime(
+        2023, 10, 25, 17, 9, 19, 736166, tzinfo=timezone.utc
+    )
+    assert result.get_datacontenttype() == "application/json"
+    assert result.get_dataschema() == "http://example.com/schema"
+    assert result.get_subject() == "test_subject"
+    assert result.get_data() is None