Update types and handle data_base64 structured.

- Add sane defaults for encoding
  - Unfortunately, defaults for structured and binary need to be *different*
   - Push types through interfaces
- Make it easy to call 'ToRequest' using Marshaller defaults
- Add tests for above

Author: evankanderson

README.md

@@ -129,7 +129,7 @@ def run_binary(event, url):
     print("binary CloudEvent")
     for k, v in binary_headers.items():
         print("{0}: {1}\r\n".format(k, v))
-    print(binary_data.getvalue())
+    print(binary_data)
     response = requests.post(url,
                              headers=binary_headers,
                              data=binary_data.getvalue())
@@ -138,14 +138,14 @@ def run_binary(event, url):
 
 def run_structured(event, url):
     structured_headers, structured_data = http_marshaller.ToRequest(
-        event, converters.TypeStructured, json.dumps
-    )
+        event, converters.TypeStructured, json.dumps)
+
     print("structured CloudEvent")
-    print(structured_data.getvalue())
+    print(structured_data.decode("utf-8"))
 
     response = requests.post(url,
                              headers=structured_headers,
-                             data=structured_data.getvalue())
+                             data=structured_data)
     response.raise_for_status()
 
 ```
@@ -163,7 +163,7 @@ The code below shows how integrate both libraries in order to create a CloudEven
     event = v02.Event()
     http_marshaller = marshaller.NewDefaultHTTPMarshaller()
     event = http_marshaller.FromRequest(
-        event, headers, data, json.load)
+        event, headers, data)
 
 ```
 Complete example of turning a CloudEvent into a request you can find [here](samples/python-requests/request_to_cloudevent.py).

cloudevents/sdk/converters/binary.py

@@ -36,16 +36,16 @@ def read(
         event: event_base.BaseEvent,
         headers: dict,
         body: typing.IO,
-        data_unmarshaller: typing.Callable,
+        data_unmarshaller: event_base.UnmarshallerType,
     ) -> event_base.BaseEvent:
         if type(event) not in self.SUPPORTED_VERSIONS:
             raise exceptions.UnsupportedEvent(type(event))
         event.UnmarshalBinary(headers, body, data_unmarshaller)
         return event
 
     def write(
-        self, event: event_base.BaseEvent, data_marshaller: typing.Callable
-    ) -> (dict, typing.IO):
+        self, event: event_base.BaseEvent, data_marshaller: event_base.MarshallerType
+    ) -> (dict, bytes):
         return event.MarshalBinary(data_marshaller)
 
 

cloudevents/sdk/converters/structured.py

@@ -35,16 +35,16 @@ def read(
         event: event_base.BaseEvent,
         headers: dict,
         body: typing.IO,
-        data_unmarshaller: typing.Callable,
+        data_unmarshaller: event_base.UnmarshallerType,
     ) -> event_base.BaseEvent:
         event.UnmarshalJSON(body, data_unmarshaller)
         return event
 
     def write(
-        self, event: event_base.BaseEvent, data_marshaller: typing.Callable
-    ) -> (dict, typing.IO):
+        self, event: event_base.BaseEvent, data_marshaller: event_base.UnmarshallerType
+    ) -> (dict, bytes):
         http_headers = {"content-type": self.MIME_TYPE}
-        return http_headers, event.MarshalJSON(data_marshaller)
+        return http_headers, event.MarshalJSON(data_marshaller).encode("utf-8")
 
 
 def NewJSONHTTPCloudEventConverter() -> JSONHTTPCloudEventConverter:

cloudevents/sdk/event/base.py

@@ -12,10 +12,16 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import base64
 import io
 import json
 import typing
 
+# Use consistent types for marshal and unmarshal functions across
+# both JSON and Binary format.
+MarshallerType = typing.Callable[[typing.Any], typing.Union[bytes, str]]
+UnmarshallerType = typing.Callable[[typing.IO], typing.Any]
+
 
 # TODO(slinkydeveloper) is this really needed?
 class EventGetterSetter(object):
@@ -110,23 +116,39 @@ def Set(self, key: str, value: object):
         exts.update({key: value})
         self.Set("extensions", exts)
 
-    def MarshalJSON(self, data_marshaller: typing.Callable) -> typing.IO:
+    def MarshalJSON(self, data_marshaller: MarshallerType) -> str:
+        if data_marshaller is None:
+            data_marshaller = lambda x: x
         props = self.Properties()
-        props["data"] = data_marshaller(props.get("data"))
-        return io.BytesIO(json.dumps(props).encode("utf-8"))
-
-    def UnmarshalJSON(self, b: typing.IO, data_unmarshaller: typing.Callable):
+        data = ""
+        if "data" in props:
+            data = data_marshaller(props.get("data"))
+            del props["data"]
+        if isinstance(data, bytes):
+            props["data_base64"] = base64.b64encode(data).decode("ascii")
+        else:
+            props["data"] = data
+        return json.dumps(props)
+
+    def UnmarshalJSON(
+        self,
+        b: typing.IO,
+        data_unmarshaller: UnmarshallerType
+    ):
         raw_ce = json.load(b)
         for name, value in raw_ce.items():
             if name == "data":
-                value = data_unmarshaller(value)
+                value = data_unmarshaller(io.StringIO(value))
+            if name == "data_base64":
+                value = data_unmarshaller(io.BytesIO(base64.b64decode(value)))
+                name = "data"
             self.Set(name, value)
 
     def UnmarshalBinary(
         self,
         headers: dict,
         body: typing.IO,
-        data_unmarshaller: typing.Callable
+        data_unmarshaller: UnmarshallerType
     ):
         for header, value in headers.items():
             header = header.lower()
@@ -139,8 +161,10 @@ def UnmarshalBinary(
 
     def MarshalBinary(
             self,
-            data_marshaller: typing.Callable
-    ) -> (dict, object):
+            data_marshaller: MarshallerType
+    ) -> (dict, bytes):
+        if data_marshaller is None:
+            data_marshaller = json.dumps
         headers = {}
         if self.ContentType():
             headers["content-type"] = self.ContentType()
@@ -154,4 +178,7 @@ def MarshalBinary(
             headers["ce-{0}".format(key)] = value
 
         data, _ = self.Get("data")
-        return headers, data_marshaller(data)
+        data = data_marshaller(data)
+        if isinstance(data, str):  # Convenience method for json.dumps
+            data = data.encode("utf-8")
+        return headers, data

cloudevents/sdk/marshaller.py

@@ -13,6 +13,7 @@
 #    under the License.
 
 import typing
+import json
 
 from cloudevents.sdk import exceptions
 
@@ -43,7 +44,7 @@ def FromRequest(
         event: event_base.BaseEvent,
         headers: dict,
         body: typing.IO,
-        data_unmarshaller: typing.Callable,
+        data_unmarshaller: event_base.UnmarshallerType = json.load,
     ) -> event_base.BaseEvent:
         """
         Reads a CloudEvent from an HTTP headers and request body
@@ -78,9 +79,9 @@ def FromRequest(
     def ToRequest(
         self,
         event: event_base.BaseEvent,
-        converter_type: str,
-        data_marshaller: typing.Callable,
-    ) -> (dict, typing.IO):
+        converter_type: str = None,
+        data_marshaller: event_base.MarshallerType = None,
+    ) -> (dict, bytes):
         """
         Writes a CloudEvent into a HTTP-ready form of headers and request body
         :param event: CloudEvent
@@ -92,9 +93,12 @@ def ToRequest(
         :return: dict of HTTP headers and stream of HTTP request body
         :rtype: tuple
         """
-        if not isinstance(data_marshaller, typing.Callable):
+        if data_marshaller is not None and not isinstance(data_marshaller, typing.Callable):
             raise exceptions.InvalidDataMarshaller()
 
+        if converter_type is None:
+            converter_type = self.__converters[0].TYPE
+
         if converter_type in self.__converters_by_type:
             cnvrtr = self.__converters_by_type[converter_type]
             return cnvrtr.write(event, data_marshaller)

cloudevents/tests/test_event_from_request_converter.py

@@ -34,7 +34,8 @@
 def test_binary_converter_upstream(event_class):
     m = marshaller.NewHTTPMarshaller(
         [binary.NewBinaryHTTPCloudEventConverter()])
-    event = m.FromRequest(event_class(), data.headers[event_class], None, lambda x: x)
+    event = m.FromRequest(
+        event_class(), data.headers[event_class], None, lambda x: x)
     assert event is not None
     assert event.EventType() == data.ce_type
     assert event.EventID() == data.ce_id
@@ -100,6 +101,7 @@ def test_structured_converter_v01():
     assert event.Get("type") == (data.ce_type, True)
     assert event.Get("id") == (data.ce_id, True)
 
+
 @pytest.mark.parametrize("event_class", [v02.Event, v03.Event, v1.Event])
 def test_default_http_marshaller_with_structured(event_class):
     m = marshaller.NewDefaultHTTPMarshaller()
@@ -147,15 +149,11 @@ def test_unsupported_event_configuration():
 
 def test_invalid_data_unmarshaller():
     m = marshaller.NewDefaultHTTPMarshaller()
-    pytest.raises(
-        exceptions.InvalidDataUnmarshaller,
-        m.FromRequest,
-        v01.Event(), {}, None, None
-    )
+    with pytest.raises(exceptions.InvalidDataUnmarshaller):
+        m.FromRequest(v01.Event(), {}, None, "string")
 
 
 def test_invalid_data_marshaller():
     m = marshaller.NewDefaultHTTPMarshaller()
-    pytest.raises(
-        exceptions.InvalidDataMarshaller, m.ToRequest, v01.Event(), "blah", None
-    )
+    with pytest.raises(exceptions.InvalidDataMarshaller):
+        m.ToRequest(v01.Event(), "blah", "string")

cloudevents/tests/test_event_pipeline.py

@@ -45,8 +45,8 @@ def test_event_pipeline_upstream(event_class):
     assert "ce-id" in new_headers
     assert "ce-time" in new_headers
     assert "content-type" in new_headers
-    assert isinstance(body, str)
-    assert data.body == body
+    assert isinstance(body, bytes)
+    assert data.body == body.decode("utf-8")
 
 
 def test_extensions_are_set_upstream():
@@ -76,8 +76,8 @@ def test_event_pipeline_v01():
     m = marshaller.NewHTTPMarshaller([structured.NewJSONHTTPCloudEventConverter()])
 
     _, body = m.ToRequest(event, converters.TypeStructured, lambda x: x)
-    assert isinstance(body, io.BytesIO)
-    new_headers = json.load(io.TextIOWrapper(body, encoding="utf-8"))
+    assert isinstance(body, bytes)
+    new_headers = json.loads(body)
     assert new_headers is not None
     assert "cloudEventsVersion" in new_headers
     assert "eventType" in new_headers
@@ -86,3 +86,41 @@ def test_event_pipeline_v01():
     assert "eventTime" in new_headers
     assert "contentType" in new_headers
     assert data.body == new_headers["data"]
+
+def test_binary_event_v1():
+    event = (
+        v1.Event()
+        .SetContentType("application/octet-stream")
+        .SetData(b'\x00\x01')
+    )
+    m = marshaller.NewHTTPMarshaller([structured.NewJSONHTTPCloudEventConverter()])
+
+    _, body = m.ToRequest(event, converters.TypeStructured, lambda x: x)
+    assert isinstance(body, bytes)
+    content = json.loads(body)
+    assert "data" not in content
+    assert content["data_base64"] == "AAE=", f"Content is: {content}"
+
+def test_object_event_v1():
+    event = (
+        v1.Event()
+        .SetContentType("application/json")
+        .SetData({"name": "john"})
+    )
+
+    m = marshaller.NewDefaultHTTPMarshaller()
+
+    _, structuredBody = m.ToRequest(event)
+    assert isinstance(structuredBody, bytes)
+    structuredObj = json.loads(structuredBody)
+    errorMsg = f"Body was {structuredBody}, obj is {structuredObj}"
+    assert isinstance(structuredObj, dict), errorMsg
+    assert isinstance(structuredObj["data"], dict), errorMsg
+    assert len(structuredObj["data"]) == 1, errorMsg
+    assert structuredObj["data"]["name"] == "john", errorMsg
+
+    headers, binaryBody = m.ToRequest(event, converters.TypeBinary)
+    assert isinstance(headers, dict)
+    assert isinstance(binaryBody, bytes)
+    assert headers["content-type"] == "application/json"
+    assert binaryBody == b'{"name": "john"}', f"Binary is {binaryBody!r}"