refactor: move all abstract conversion logic under conversion

Signed-off-by: Alexander Tkachev <sasha64sasha@gmail.com>
This commit is contained in:
Alexander Tkachev 2022-07-25 00:48:37 +03:00
parent f47087d490
commit 41c5f5984b
8 changed files with 56 additions and 74 deletions

View File

@ -12,4 +12,4 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
from cloudevents.abstract.event import CloudEvent, AnyCloudEvent # noqa from cloudevents.abstract.event import AnyCloudEvent, CloudEvent # noqa

View File

@ -1,4 +1,3 @@
import abc
import typing import typing
from typing import TypeVar from typing import TypeVar

View File

@ -1,54 +0,0 @@
# Copyright 2018-Present The CloudEvents Authors
#
# 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.
import typing
from cloudevents.abstract import AnyCloudEvent
from cloudevents.sdk import types
from cloudevents.abstract.http_methods import to_structured, from_http
def to_json(
event: AnyCloudEvent,
data_marshaller: types.MarshallerType = None,
) -> typing.Union[str, bytes]:
"""
Cast an CloudEvent into a json object
:param event: CloudEvent which will be converted into a json object
:type event: CloudEvent
:param data_marshaller: Callable function which will cast event.data
into a json object
:type data_marshaller: typing.Callable
:returns: json object representing the given event
"""
return to_structured(event, data_marshaller=data_marshaller)[1]
def from_json(
event_type: typing.Type[AnyCloudEvent],
data: typing.Union[str, bytes],
data_unmarshaller: types.UnmarshallerType = None,
) -> AnyCloudEvent:
"""
Cast json encoded data into an CloudEvent
:param event_type: Concrete event type to which deserialize the json event
:param data: json encoded cloudevent data
:param data_unmarshaller: Callable function which will cast data to a
python object
:type data_unmarshaller: typing.Callable
:returns: CloudEvent representing given cloudevent json object
"""
return from_http(
event_type, headers={}, data=data, data_unmarshaller=data_unmarshaller
)

View File

@ -11,19 +11,52 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import json import json
import typing import typing
from cloudevents import exceptions as cloud_exceptions
import cloudevents.exceptions as cloud_exceptions from cloudevents.abstract import AnyCloudEvent
from cloudevents.abstract.event import AnyCloudEvent from cloudevents.http import is_binary
from cloudevents.http.event_type import is_binary
from cloudevents.http.mappings import _marshaller_by_format, _obj_by_version from cloudevents.http.mappings import _marshaller_by_format, _obj_by_version
from cloudevents.http.util import _json_or_string from cloudevents.http.util import _json_or_string
from cloudevents.sdk import converters, marshaller, types from cloudevents.sdk import converters, marshaller, types
def to_json(
event: AnyCloudEvent,
data_marshaller: types.MarshallerType = None,
) -> typing.Union[str, bytes]:
"""
Cast an CloudEvent into a json object
:param event: CloudEvent which will be converted into a json object
:type event: CloudEvent
:param data_marshaller: Callable function which will cast event.data
into a json object
:type data_marshaller: typing.Callable
:returns: json object representing the given event
"""
return to_structured(event, data_marshaller=data_marshaller)[1]
def from_json(
event_type: typing.Type[AnyCloudEvent],
data: typing.Union[str, bytes],
data_unmarshaller: types.UnmarshallerType = None,
) -> AnyCloudEvent:
"""
Cast json encoded data into an CloudEvent
:param event_type: Concrete event type to which deserialize the json event
:param data: json encoded cloudevent data
:param data_unmarshaller: Callable function which will cast data to a
python object
:type data_unmarshaller: typing.Callable
:returns: CloudEvent representing given cloudevent json object
"""
return from_http(
event_type, headers={}, data=data, data_unmarshaller=data_unmarshaller
)
def from_http( def from_http(
event_type: typing.Type[AnyCloudEvent], event_type: typing.Type[AnyCloudEvent],
headers: typing.Dict[str, str], headers: typing.Dict[str, str],

View File

@ -17,8 +17,8 @@ import typing
import uuid import uuid
import cloudevents.exceptions as cloud_exceptions import cloudevents.exceptions as cloud_exceptions
from cloudevents.http.mappings import _required_by_version
from cloudevents import abstract from cloudevents import abstract
from cloudevents.http.mappings import _required_by_version
class CloudEvent(abstract.CloudEvent): class CloudEvent(abstract.CloudEvent):

View File

@ -12,20 +12,15 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import json
import typing import typing
from deprecation import deprecated from deprecation import deprecated
from cloudevents.conversion import from_http as _abstract_from_http
from cloudevents.conversion import to_binary, to_structured
from cloudevents.http.event import CloudEvent from cloudevents.http.event import CloudEvent
from cloudevents.sdk import types from cloudevents.sdk import types
# backwards compatability imports
from cloudevents.abstract.http_methods import to_binary, to_structured # noqa
from cloudevents.abstract.http_methods import from_http as _abstract_from_http
def from_http( def from_http(
headers: typing.Dict[str, str], headers: typing.Dict[str, str],
@ -46,6 +41,12 @@ def from_http(
return _abstract_from_http(CloudEvent, headers, data, data_unmarshaller) return _abstract_from_http(CloudEvent, headers, data, data_unmarshaller)
# backwards compatibility
to_binary = to_binary
# backwards compatibility
to_structured = to_structured
@deprecated(deprecated_in="1.0.2", details="Use to_binary function instead") @deprecated(deprecated_in="1.0.2", details="Use to_binary function instead")
def to_binary_http( def to_binary_http(
event: CloudEvent, data_marshaller: types.MarshallerType = None event: CloudEvent, data_marshaller: types.MarshallerType = None

View File

@ -14,13 +14,11 @@
import typing import typing
from cloudevents.conversion import from_json as _abstract_from_json
from cloudevents.conversion import to_json
from cloudevents.http.event import CloudEvent from cloudevents.http.event import CloudEvent
from cloudevents.sdk import types from cloudevents.sdk import types
# backwards compatibility
from cloudevents.abstract.json_methods import to_json # noqa
from cloudevents.abstract.json_methods import from_json as _abstract_from_json
def from_json( def from_json(
data: typing.Union[str, bytes], data: typing.Union[str, bytes],
@ -35,3 +33,7 @@ def from_json(
:returns: CloudEvent representing given cloudevent json object :returns: CloudEvent representing given cloudevent json object
""" """
return _abstract_from_json(CloudEvent, data, data_unmarshaller) return _abstract_from_json(CloudEvent, data, data_unmarshaller)
# backwards compatibility
to_json = to_json

View File

@ -1,6 +1,7 @@
from cloudevents.abstract import CloudEvent
import pytest import pytest
from cloudevents.abstract import CloudEvent
def test_del_is_abstract(): def test_del_is_abstract():
""" """