cloudevent fields type checking adjustments

cloudevents/exceptions.py

@@ -0,0 +1,19 @@
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+class CloudEventMissingRequiredFields(Exception):
+    pass
+
+
+class CloudEventTypeErrorRequiredFields(Exception):
+    pass

cloudevents/http/__init__.py

@@ -15,6 +15,7 @@
 import typing
 
 from cloudevents.http.event import CloudEvent
+from cloudevents.http.event_type import is_binary, is_structured
 from cloudevents.http.http_methods import (
     from_http,
     to_binary_http,

cloudevents/http/event.py

@@ -16,6 +16,7 @@
 import typing
 import uuid
 
+import cloudevents.exceptions as cloud_exceptions
 from cloudevents.http.mappings import _required_by_version
 
 
@@ -57,14 +58,14 @@ def __init__(
             ).isoformat()
 
         if self._attributes["specversion"] not in _required_by_version:
-            raise ValueError(
+            raise cloud_exceptions.CloudEventMissingRequiredFields(
                 f"Invalid specversion: {self._attributes['specversion']}"
             )
         # There is no good way to default 'source' and 'type', so this
         # checks for those (or any new required attributes).
         required_set = _required_by_version[self._attributes["specversion"]]
         if not required_set <= self._attributes.keys():
-            raise ValueError(
+            raise cloud_exceptions.CloudEventMissingRequiredFields(
                 f"Missing required keys: {required_set - attributes.keys()}"
             )
 

cloudevents/http/event_type.py

@@ -0,0 +1,29 @@
+import typing
+
+from cloudevents.sdk.converters import binary, structured
+
+
+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/http/http_methods.py

@@ -1,7 +1,9 @@
 import json
 import typing
 
+import cloudevents.exceptions as cloud_exceptions
 from cloudevents.http.event import CloudEvent
+from cloudevents.http.event_type import is_binary, is_structured
 from cloudevents.http.mappings import _marshaller_by_format, _obj_by_version
 from cloudevents.http.util import _json_or_string
 from cloudevents.sdk import converters, marshaller, types
@@ -27,19 +29,23 @@ def from_http(
 
     marshall = marshaller.NewDefaultHTTPMarshaller()
 
-    if converters.is_binary(headers):
+    if 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")
+        raise cloud_exceptions.CloudEventMissingRequiredFields(
+            "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}")
+        raise cloud_exceptions.CloudEventTypeErrorRequiredFields(
+            f"found invalid specversion {specversion}"
+        )
 
     event = marshall.FromRequest(
         event_handler(), headers, data, data_unmarshaller=data_unmarshaller
@@ -71,7 +77,7 @@ def _to_http(
         data_marshaller = _marshaller_by_format[format]
 
     if event._attributes["specversion"] not in _obj_by_version:
-        raise ValueError(
+        raise cloud_exceptions.CloudEventTypeErrorRequiredFields(
             f"Unsupported specversion: {event._attributes['specversion']}"
         )
 

cloudevents/sdk/converters/__init__.py

@@ -11,36 +11,7 @@
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    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,7 +16,7 @@
 
 from cloudevents.sdk import exceptions, types
 from cloudevents.sdk.converters import base
-from cloudevents.sdk.converters.structured import JSONHTTPCloudEventConverter
+from cloudevents.sdk.converters.util import has_binary_headers
 from cloudevents.sdk.event import base as event_base
 from cloudevents.sdk.event import v1, v03
 
@@ -28,13 +28,11 @@ class BinaryHTTPCloudEventConverter(base.Converter):
 
     def can_read(
         self,
-        content_type: str,
+        content_type: str = None,
         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)
-        )
+
+        return has_binary_headers(headers)
 
     def event_supported(self, event: object) -> bool:
         return type(event) in self.SUPPORTED_VERSIONS

cloudevents/sdk/converters/structured.py

@@ -16,23 +16,25 @@
 
 from cloudevents.sdk import types
 from cloudevents.sdk.converters import base
+from cloudevents.sdk.converters.binary import BinaryHTTPCloudEventConverter
+from cloudevents.sdk.converters.util import has_binary_headers
 from cloudevents.sdk.event import base as event_base
 
 
+# TODO: Singleton?
 class JSONHTTPCloudEventConverter(base.Converter):
 
     TYPE = "structured"
     MIME_TYPE = "application/cloudevents+json"
 
     def can_read(
-        self,
-        content_type: str,
-        headers: typing.Dict[str, str] = {"ce-specversion": None},
+        self, content_type: str, headers: typing.Dict[str, str] = {},
     ) -> bool:
         return (
             isinstance(content_type, str)
             and content_type.startswith(self.MIME_TYPE)
-        ) or ("ce-specversion" not in headers)
+            or not has_binary_headers(headers)
+        )
 
     def event_supported(self, event: object) -> bool:
         # structured format supported by both spec 0.1 and 0.2

cloudevents/sdk/converters/util.py

@@ -0,0 +1,10 @@
+import typing
+
+
+def has_binary_headers(headers: typing.Dict[str, str]) -> bool:
+    return (
+        "ce-specversion" in headers
+        and "ce-source" in headers
+        and "ce-type" in headers
+        and "ce-id" in headers
+    )

cloudevents/sdk/event/base.py

@@ -16,10 +16,12 @@
 import json
 import typing
 
+import cloudevents.exceptions as cloud_exceptions
 from cloudevents.sdk import types
 
-
 # TODO(slinkydeveloper) is this really needed?
+
+
 class EventGetterSetter(object):
 
     # ce-specversion
@@ -159,6 +161,9 @@ def content_type(self, value: str):
 
 
 class BaseEvent(EventGetterSetter):
+    _ce_required_fields = set()
+    _ce_optional_fields = set()
+
     def Properties(self, with_nullable=False) -> dict:
         props = dict()
         for name, value in self.__dict__.items():
@@ -193,7 +198,10 @@ def Set(self, key: str, value: object):
 
     def MarshalJSON(self, data_marshaller: types.MarshallerType) -> str:
         if data_marshaller is None:
-            data_marshaller = lambda x: x  # noqa: E731
+
+            def data_marshaller(x):
+                return x  # noqa: E731
+
         props = self.Properties()
         if "data" in props:
             data = data_marshaller(props.pop("data"))
@@ -215,7 +223,9 @@ def UnmarshalJSON(
 
         missing_fields = self._ce_required_fields - raw_ce.keys()
         if len(missing_fields) > 0:
-            raise ValueError(f"Missing required attributes: {missing_fields}")
+            raise cloud_exceptions.CloudEventMissingRequiredFields(
+                f"Missing required attributes: {missing_fields}"
+            )
 
         for name, value in raw_ce.items():
             if name == "data":
@@ -233,18 +243,23 @@ def UnmarshalBinary(
         body: typing.Union[bytes, str],
         data_unmarshaller: types.UnmarshallerType,
     ):
-        if "ce-specversion" not in headers:
-            raise ValueError("Missing required attribute: 'specversion'")
+        required_binary_fields = {
+            f"ce-{field}" for field in self._ce_required_fields
+        }
+        missing_fields = required_binary_fields - headers.keys()
+
+        if len(missing_fields) > 0:
+            raise cloud_exceptions.CloudEventMissingRequiredFields(
+                f"Missing required attributes: {missing_fields}"
+            )
+
         for header, value in headers.items():
             header = header.lower()
             if header == "content-type":
                 self.SetContentType(value)
             elif header.startswith("ce-"):
                 self.Set(header[3:], value)
         self.Set("data", data_unmarshaller(body))
-        missing_attrs = self._ce_required_fields - self.Properties().keys()
-        if len(missing_attrs) > 0:
-            raise ValueError(f"Missing required attributes: {missing_attrs}")
 
     def MarshalBinary(
         self, data_marshaller: types.MarshallerType