Specversion toggling

cloudevents/sdk/converters/__init__.py

@@ -12,7 +12,35 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import typing
+
 from cloudevents.sdk.converters import binary, structured
 
 TypeBinary = binary.BinaryHTTPCloudEventConverter.TYPE
 TypeStructured = structured.JSONHTTPCloudEventConverter.TYPE
+
+
+def is_binary(headers: typing.Dict[str, str]) -> bool:
+    """Uses internal marshallers to determine whether this event is binary
+    :param headers: the HTTP headers
+    :type headers: typing.Dict[str, str]
+    :returns bool: returns a bool indicating whether the headers indicate a binary event type
+    """
+    headers = {key.lower(): value for key, value in headers.items()}
+    content_type = headers.get("content-type", "")
+    binary_parser = binary.BinaryHTTPCloudEventConverter()
+    return binary_parser.can_read(content_type=content_type, headers=headers)
+
+
+def is_structured(headers: typing.Dict[str, str]) -> bool:
+    """Uses internal marshallers to determine whether this event is structured
+    :param headers: the HTTP headers
+    :type headers: typing.Dict[str, str]
+    :returns bool: returns a bool indicating whether the headers indicate a structured event type
+    """
+    headers = {key.lower(): value for key, value in headers.items()}
+    content_type = headers.get("content-type", "")
+    structured_parser = structured.JSONHTTPCloudEventConverter()
+    return structured_parser.can_read(
+        content_type=content_type, headers=headers
+    )

cloudevents/sdk/converters/binary.py

@@ -16,6 +16,7 @@
 
 from cloudevents.sdk import exceptions, types
 from cloudevents.sdk.converters import base
+from cloudevents.sdk.converters.structured import JSONHTTPCloudEventConverter
 from cloudevents.sdk.event import base as event_base
 from cloudevents.sdk.event import v1, v03
 
@@ -25,8 +26,15 @@ class BinaryHTTPCloudEventConverter(base.Converter):
     TYPE = "binary"
     SUPPORTED_VERSIONS = [v03.Event, v1.Event]
 
-    def can_read(self, content_type: str) -> bool:
-        return True
+    def can_read(
+        self,
+        content_type: str,
+        headers: typing.Dict[str, str] = {"ce-specversion": None},
+    ) -> bool:
+        return ("ce-specversion" in headers) and not (
+            isinstance(content_type, str)
+            and content_type.startswith(JSONHTTPCloudEventConverter.MIME_TYPE)
+        )
 
     def event_supported(self, event: object) -> bool:
         return type(event) in self.SUPPORTED_VERSIONS

cloudevents/sdk/converters/structured.py

@@ -24,8 +24,15 @@ class JSONHTTPCloudEventConverter(base.Converter):
     TYPE = "structured"
     MIME_TYPE = "application/cloudevents+json"
 
-    def can_read(self, content_type: str) -> bool:
-        return content_type and content_type.startswith(self.MIME_TYPE)
+    def can_read(
+        self,
+        content_type: str,
+        headers: typing.Dict[str, str] = {"ce-specversion": None},
+    ) -> bool:
+        return (
+            isinstance(content_type, str)
+            and content_type.startswith(self.MIME_TYPE)
+        ) or ("ce-specversion" not in headers)
 
     def event_supported(self, event: object) -> bool:
         # structured format supported by both spec 0.1 and 0.2

cloudevents/sdk/http/__init__.py

@@ -18,6 +18,7 @@
 from cloudevents.sdk.event import v1, v03
 from cloudevents.sdk.http.event import (
     EventClass,
+    _obj_by_version,
     to_binary_http,
     to_structured_http,
 )
@@ -41,21 +42,36 @@ def from_http(
     data: typing.Union[str, bytes],
     headers: typing.Dict[str, str],
     data_unmarshaller: types.UnmarshallerType = None,
-) -> CloudEvent:
-
+):
     """Unwrap a CloudEvent (binary or structured) from an HTTP request.
-        :param data: the HTTP request body
-        :type data: typing.IO
-        :param headers: the HTTP headers
-        :type headers: typing.Dict[str, str]
-        :param data_unmarshaller: Function to decode data into a python object.
-        :type data_unmarshaller: types.UnmarshallerType
-        """
+    :param data: the HTTP request body
+    :type data: typing.IO
+    :param headers: the HTTP headers
+    :type headers: typing.Dict[str, str]
+    :param data_unmarshaller: Function to decode data into a python object.
+    :type data_unmarshaller: types.UnmarshallerType
+    """
     if data_unmarshaller is None:
         data_unmarshaller = _json_or_string
 
-    event = marshaller.NewDefaultHTTPMarshaller().FromRequest(
-        v1.Event(), headers, data, data_unmarshaller
+    marshall = marshaller.NewDefaultHTTPMarshaller()
+
+    if converters.is_binary(headers):
+        specversion = headers.get("ce-specversion", None)
+    else:
+        raw_ce = json.loads(data)
+        specversion = raw_ce.get("specversion", None)
+
+    if specversion is None:
+        raise ValueError("could not find specversion in HTTP request")
+
+    event_handler = _obj_by_version.get(specversion, None)
+
+    if event_handler is None:
+        raise ValueError(f"found invalid specversion {specversion}")
+
+    event = marshall.FromRequest(
+        event_handler(), headers, data, data_unmarshaller
     )
     attrs = event.Properties()
     attrs.pop("data", None)

cloudevents/sdk/http/event.py

@@ -18,7 +18,9 @@
 import uuid
 
 from cloudevents.sdk import converters, marshaller, types
+from cloudevents.sdk.converters import is_binary
 from cloudevents.sdk.event import v1, v03
+from cloudevents.sdk.marshaller import HTTPMarshaller
 
 _marshaller_by_format = {
     converters.TypeStructured: lambda x: x,

cloudevents/sdk/marshaller.py

@@ -32,8 +32,8 @@ def __init__(self, converters: typing.List[base.Converter]):
         :param converters: a list of HTTP-to-CloudEvent-to-HTTP constructors
         :type converters: typing.List[base.Converter]
         """
-        self.__converters = [c for c in converters]
-        self.__converters_by_type = {c.TYPE: c for c in converters}
+        self.http_converters = [c for c in converters]
+        self.http_converters_by_type = {c.TYPE: c for c in converters}
 
     def FromRequest(
         self,
@@ -62,13 +62,15 @@ def FromRequest(
         headers = {key.lower(): value for key, value in headers.items()}
         content_type = headers.get("content-type", None)
 
-        for cnvrtr in self.__converters:
-            if cnvrtr.can_read(content_type) and cnvrtr.event_supported(event):
+        for cnvrtr in self.http_converters:
+            if cnvrtr.can_read(
+                content_type, headers=headers
+            ) and cnvrtr.event_supported(event):
                 return cnvrtr.read(event, headers, body, data_unmarshaller)
 
         raise exceptions.UnsupportedEventConverter(
             "No registered marshaller for {0} in {1}".format(
-                content_type, self.__converters
+                content_type, self.http_converters
             )
         )
 
@@ -95,10 +97,10 @@ def ToRequest(
             raise exceptions.InvalidDataMarshaller()
 
         if converter_type is None:
-            converter_type = self.__converters[0].TYPE
+            converter_type = self.http_converters[0].TYPE
 
-        if converter_type in self.__converters_by_type:
-            cnvrtr = self.__converters_by_type[converter_type]
+        if converter_type in self.http_converters_by_type:
+            cnvrtr = self.http_converters_by_type[converter_type]
             return cnvrtr.write(event, data_marshaller)
 
         raise exceptions.NoSuchConverter(converter_type)

cloudevents/tests/test_http_events.py

@@ -206,7 +206,7 @@ def test_missing_ce_prefix_binary_event(specversion):
             # and NotImplementedError because structured calls aren't
             # implemented. In this instance one of the required keys should have
             # prefix e-id instead of ce-id therefore it should throw
-            _ = from_http(test_data, headers=prefixed_headers)
+            _ = from_http(json.dumps(test_data), headers=prefixed_headers)
 
 
 @pytest.mark.parametrize("specversion", ["1.0", "0.3"])
@@ -332,6 +332,60 @@ def test_valid_structured_events(specversion):
         assert event.data["payload"] == f"payload-{i}"
 
 
+@pytest.mark.parametrize("specversion", ["1.0", "0.3"])
+def test_structured_no_content_type(specversion):
+    # Test creating multiple cloud events
+    events_queue = []
+    headers = {}
+    num_cloudevents = 30
+    data = {
+        "id": "id",
+        "source": "source.com.test",
+        "type": "cloudevent.test.type",
+        "specversion": specversion,
+        "data": test_data,
+    }
+    event = from_http(json.dumps(data), {},)
+
+    assert event["id"] == "id"
+    assert event["source"] == "source.com.test"
+    assert event["specversion"] == specversion
+    for key, val in test_data.items():
+        assert event.data[key] == val
+
+
+def test_is_binary():
+    headers = {
+        "ce-id": "my-id",
+        "ce-source": "<event-source>",
+        "ce-type": "cloudevent.event.type",
+        "ce-specversion": "1.0",
+        "Content-Type": "text/plain",
+    }
+    assert converters.is_binary(headers)
+
+    headers = {
+        "Content-Type": "application/cloudevents+json",
+    }
+    assert not converters.is_binary(headers)
+
+    headers = {}
+    assert not converters.is_binary(headers)
+
+
+def test_is_structured():
+    headers = {
+        "Content-Type": "application/cloudevents+json",
+    }
+    assert converters.is_structured(headers)
+
+    headers = {}
+    assert converters.is_structured(headers)
+
+    headers = {"ce-specversion": "1.0"}
+    assert not converters.is_structured(headers)
+
+
 @pytest.mark.parametrize("specversion", ["1.0", "0.3"])
 def test_cloudevent_repr(specversion):
     headers = {