feat: pydantic

cloudevents/conversion.py

@@ -23,15 +23,44 @@
 #    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 datetime
+import enum
 import json
 import typing
 
 from cloudevents import exceptions as cloud_exceptions
 from cloudevents.abstract import AnyCloudEvent
-from cloudevents.http import is_binary
-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
+from cloudevents.sdk.converters import is_binary
+from cloudevents.sdk.event import v1, v03
+
+
+def _best_effort_serialize_to_json(
+    value: typing.Any, *args, **kwargs
+) -> typing.Optional[typing.Union[bytes, str, typing.Any]]:
+    """
+    Serializes the given value into a JSON-encoded string.
+
+    Given a None value returns None as is.
+    Given a non-JSON-serializable value returns return the value as is.
+
+    :param value:  The value to be serialized into a JSON string.
+    :return: JSON string of the given value OR None OR given value.
+    """
+    if value is None:
+        return None
+    try:
+        return json.dumps(value, *args, **kwargs)
+    except TypeError:
+        return value
+
+
+_default_marshaller_by_format = {
+    converters.TypeStructured: lambda x: x,
+    converters.TypeBinary: _best_effort_serialize_to_json,
+}  # type: typing.Dict[str, types.MarshallerType]
+
+_obj_by_version = {"1.0": v1.Event, "0.3": v03.Event}
 
 
 def to_json(
@@ -169,7 +198,7 @@ def _to_http(
     :returns: (http_headers: dict, http_body: bytes or str)
     """
     if data_marshaller is None:
-        data_marshaller = _marshaller_by_format[format]
+        data_marshaller = _default_marshaller_by_format[format]
 
     if event["specversion"] not in _obj_by_version:
         raise cloud_exceptions.InvalidRequiredFields(
@@ -222,3 +251,76 @@ def to_binary(
         format=converters.TypeBinary,
         data_marshaller=data_marshaller,
     )
+
+
+def best_effort_encode_attribute_value(value: typing.Any) -> typing.Any:
+    """
+    SHOULD convert any value into a JSON serialization friendly format.
+
+    This function acts in a best-effort manner and MAY not actually encode the value
+    if it does not know how to do that, or the value is already JSON-friendly.
+
+    :param value: Value which MAY or MAY NOT be JSON serializable.
+    :return: Possibly encoded value.
+    """
+    if isinstance(value, enum.Enum):
+        return value.value
+    if isinstance(value, datetime.datetime):
+        return value.isoformat()
+
+    return value
+
+
+def from_dict(
+    event_type: typing.Type[AnyCloudEvent],
+    event: typing.Dict[str, typing.Any],
+) -> AnyCloudEvent:
+    """
+    Constructs an Event object of a given `event_type` from
+    a dict `event` representation.
+
+    :param event: The event represented as a  dict.
+    :param event_type: The type of the event to be constructed from the dict.
+    :returns: The event of the specified type backed by the given dict.
+    """
+    attributes = {
+        attr_name: best_effort_encode_attribute_value(attr_value)
+        for attr_name, attr_value in event.items()
+        if attr_name != "data"
+    }
+    return event_type.create(attributes=attributes, data=event.get("data"))
+
+
+def to_dict(event: AnyCloudEvent) -> typing.Dict[str, typing.Any]:
+    """
+    Converts given `event` to its canonical dictionary representation.
+
+    :param event: The event to be converted into a dict.
+    :returns: The canonical dict representation of the event.
+    """
+    result = {attribute_name: event.get(attribute_name) for attribute_name in event}
+    result["data"] = event.data
+    return result
+
+
+def _json_or_string(
+    content: typing.Optional[typing.AnyStr],
+) -> typing.Optional[
+    typing.Union[
+        typing.Dict[typing.Any, typing.Any],
+        typing.List[typing.Any],
+        typing.AnyStr,
+    ]
+]:
+    """
+    Returns a JSON-decoded dictionary or a list of dictionaries if
+    a valid JSON string is provided.
+
+    Returns the same `content` in case of an error or `None` when no content provided.
+    """
+    if content is None:
+        return None
+    try:
+        return json.loads(content)
+    except (json.JSONDecodeError, TypeError, UnicodeDecodeError):
+        return content

cloudevents/exceptions.py

@@ -39,3 +39,16 @@ class DataMarshallerError(GenericException):
 
 class DataUnmarshallerError(GenericException):
     pass
+
+
+class IncompatibleArgumentsError(GenericException):
+    """
+    Raised when a user tries to call a function with arguments which are incompatible
+    with each other.
+    """
+
+
+class PydanticFeatureNotInstalled(GenericException):
+    """
+    Raised when a user tries to use the pydantic feature but did not install it.
+    """

cloudevents/http/__init__.py

@@ -12,13 +12,18 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-from cloudevents.http.event import CloudEvent  # noqa
-from cloudevents.http.event_type import is_binary, is_structured  # noqa
-from cloudevents.http.http_methods import (  # noqa
+
+from cloudevents.http.conversion import (  # noqa
+    from_dict,
     from_http,
+    from_json,
     to_binary,
-    to_binary_http,
+    to_dict,
+    to_json,
     to_structured,
-    to_structured_http,
 )
-from cloudevents.http.json_methods import from_json, to_json  # noqa
+from cloudevents.http.event import CloudEvent  # noqa
+from cloudevents.http.http_methods import to_binary_http  # deprecated # noqa
+from cloudevents.http.http_methods import to_structured_http  # deprecated # noqa
+from cloudevents.sdk.converters.binary import is_binary  # noqa
+from cloudevents.sdk.converters.structured import is_structured  # noqa

cloudevents/http/conversion.py

@@ -0,0 +1,56 @@
+import typing
+
+from cloudevents.conversion import from_dict as _abstract_from_dict
+from cloudevents.conversion import from_http as _abstract_from_http
+from cloudevents.conversion import from_json as _abstract_from_json
+from cloudevents.conversion import to_binary, to_dict, to_json, to_structured  # noqa
+from cloudevents.http.event import CloudEvent
+from cloudevents.sdk import types
+
+
+def from_json(
+    data: typing.Union[str, bytes],
+    data_unmarshaller: types.UnmarshallerType = None,
+) -> CloudEvent:
+    """
+    Parses JSON string `data` into a CloudEvent.
+
+    :param data: JSON string representation of a CloudEvent.
+    :param data_unmarshaller: Callable function that casts `data` to a
+        Python object.
+    :returns: A CloudEvent parsed from the given JSON representation.
+    """
+    return _abstract_from_json(CloudEvent, data, data_unmarshaller)
+
+
+def from_http(
+    headers: typing.Dict[str, str],
+    data: typing.Union[str, bytes, None],
+    data_unmarshaller: types.UnmarshallerType = None,
+) -> CloudEvent:
+    """
+    Parses CloudEvent `data` and `headers` into a CloudEvent`.
+
+    The method supports both binary and structured representations.
+
+    :param headers: The HTTP request headers.
+    :param data: The HTTP request body. If set to None, "" or b'', the returned
+        event's `data` field will be set to None.
+    :param data_unmarshaller: Callable function to map data to a python object
+        e.g. lambda x: x or lambda x: json.loads(x)
+    :returns: A CloudEvent instance parsed from the passed HTTP parameters of
+        the specified type.
+    """
+    return _abstract_from_http(CloudEvent, headers, data, data_unmarshaller)
+
+
+def from_dict(
+    event: typing.Dict[str, typing.Any],
+) -> CloudEvent:
+    """
+    Constructs a CloudEvent from a dict `event` representation.
+
+    :param event: The event represented as a dict.
+    :returns: The event of the specified type backed by the given dict.
+    """
+    return _abstract_from_dict(CloudEvent, event)

cloudevents/http/event.py

@@ -18,7 +18,12 @@
 
 import cloudevents.exceptions as cloud_exceptions
 from cloudevents import abstract
-from cloudevents.http.mappings import _required_by_version
+from cloudevents.sdk.event import v1, v03
+
+_required_by_version = {
+    "1.0": v1.Event._ce_required_fields,
+    "0.3": v03.Event._ce_required_fields,
+}
 
 
 class CloudEvent(abstract.CloudEvent):
@@ -41,11 +46,11 @@ def __init__(self, attributes: typing.Dict[str, str], data: typing.Any = None):
             attributes 'specversion', 'id' or 'time', this will create
             those attributes with default values.
             e.g. {
-                "content-type": "application/cloudevents+json",
-                "id": "16fb5f0b-211e-1102-3dfe-ea6e2806f124",
-                "source": "<event-source>",
-                "type": "cloudevent.event.type",
-                "specversion": "0.2"
+                "specversion": "1.0",
+                "type": "com.github.pull_request.opened",
+                "source": "https://github.com/cloudevents/spec/pull",
+                "id": "A234-1234-1234",
+                "time": "2018-04-05T17:31:00Z",
             }
         :type attributes: typing.Dict[str, str]
         :param data: The payload of the event, as a python object

cloudevents/http/event_type.py

@@ -11,33 +11,27 @@
 #    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
+from deprecation import deprecated
+
+from cloudevents.sdk.converters import is_binary as _moved_is_binary
+from cloudevents.sdk.converters import is_structured as _moved_is_structured
+
+# THIS MODULE IS DEPRECATED, YOU SHOULD NOT ADD NEW FUNCTIONALLY HERE
 
 
+@deprecated(
+    deprecated_in="1.6.0",
+    details="Use cloudevents.sdk.converters.is_binary function instead",
+)
 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)
+    return _moved_is_binary(headers)
 
 
+@deprecated(
+    deprecated_in="1.6.0",
+    details="Use cloudevents.sdk.converters.is_structured function instead",
+)
 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)
+    return _moved_is_structured(headers)

cloudevents/http/http_methods.py

@@ -16,46 +16,58 @@
 
 from deprecation import deprecated
 
-from cloudevents.conversion import from_http as _abstract_from_http
-from cloudevents.conversion import to_binary, to_structured
+from cloudevents.abstract import AnyCloudEvent
+from cloudevents.http.conversion import from_http as _moved_from_http
+from cloudevents.http.conversion import to_binary as _moved_to_binary
+from cloudevents.http.conversion import to_structured as _moved_to_structured
 from cloudevents.http.event import CloudEvent
 from cloudevents.sdk import types
 
+# THIS MODULE IS DEPRECATED, YOU SHOULD NOT ADD NEW FUNCTIONALLY HERE
 
+
+@deprecated(
+    deprecated_in="1.6.0",
+    details="Use cloudevents.http.to_binary function instead",
+)
+def to_binary(
+    event: AnyCloudEvent, data_marshaller: types.MarshallerType = None
+) -> typing.Tuple[dict, typing.Union[bytes, str]]:
+    return _moved_to_binary(event, data_marshaller)
+
+
+@deprecated(
+    deprecated_in="1.6.0",
+    details="Use cloudevents.http.to_structured function instead",
+)
+def to_structured(
+    event: AnyCloudEvent,
+    data_marshaller: types.MarshallerType = None,
+) -> typing.Tuple[dict, typing.Union[bytes, str]]:
+    return _moved_to_structured(event, data_marshaller)
+
+
+@deprecated(
+    deprecated_in="1.6.0",
+    details="Use cloudevents.http.from_http function instead",
+)
 def from_http(
     headers: typing.Dict[str, str],
     data: typing.Union[str, bytes, None],
     data_unmarshaller: types.UnmarshallerType = None,
 ) -> CloudEvent:
-    """
-    Unwrap a CloudEvent (binary or structured) from an HTTP request.
-    :param headers: the HTTP headers
-    :type headers: typing.Dict[str, str]
-    :param data: the HTTP request body. If set to None, "" or b'', the returned
-        event's data field will be set to None
-    :type data: typing.IO
-    :param data_unmarshaller: Callable function to map data to a python object
-        e.g. lambda x: x or lambda x: json.loads(x)
-    :type data_unmarshaller: types.UnmarshallerType
-    """
-    return _abstract_from_http(CloudEvent, headers, data, data_unmarshaller)
-
-
-# backwards compatibility
-to_binary = to_binary
-# backwards compatibility
-to_structured = to_structured
+    return _moved_from_http(headers, data, data_unmarshaller)
 
 
 @deprecated(deprecated_in="1.0.2", details="Use to_binary function instead")
 def to_binary_http(
     event: CloudEvent, data_marshaller: types.MarshallerType = None
 ) -> typing.Tuple[dict, typing.Union[bytes, str]]:
-    return to_binary(event, data_marshaller)
+    return _moved_to_binary(event, data_marshaller)
 
 
 @deprecated(deprecated_in="1.0.2", details="Use to_structured function instead")
 def to_structured_http(
     event: CloudEvent, data_marshaller: types.MarshallerType = None
 ) -> typing.Tuple[dict, typing.Union[bytes, str]]:
-    return to_structured(event, data_marshaller)
+    return _moved_to_structured(event, data_marshaller)