# Path: cloudevents/sdk/types.py

# 
# Path: cloudevents/sdk/converters/base.py
# class Converter(object):
#     TYPE = None
#     def read(
#         self,
#         event,
#         headers: dict,
#         body: typing.IO,
#         data_unmarshaller: typing.Callable,
#     ) -> base.BaseEvent:
#     def event_supported(self, event: object) -> bool:
#     def can_read(self, content_type: str) -> bool:
#     def write(
#         self, event: base.BaseEvent, data_marshaller: typing.Callable
#     ) -> (dict, object):
# 
# Path: cloudevents/sdk/converters/util.py
# 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
#     )
# 
# Path: cloudevents/sdk/event/base.py
# class EventGetterSetter(object):  # pragma: no cover
# class BaseEvent(EventGetterSetter):
#     def CloudEventVersion(self) -> str:
#     def specversion(self):
#     def SetCloudEventVersion(self, specversion: str) -> object:
#     def specversion(self, value: str):
#     def EventType(self) -> str:
#     def type(self):
#     def SetEventType(self, eventType: str) -> object:
#     def type(self, value: str):
#     def Source(self) -> str:
#     def source(self):
#     def SetSource(self, source: str) -> object:
#     def source(self, value: str):
#     def EventID(self) -> str:
#     def id(self):
#     def SetEventID(self, eventID: str) -> object:
#     def id(self, value: str):
#     def EventTime(self) -> str:
#     def time(self):
#     def SetEventTime(self, eventTime: str) -> object:
#     def time(self, value: str):
#     def SchemaURL(self) -> str:
#     def schema(self) -> str:
#     def SetSchemaURL(self, schemaURL: str) -> object:
#     def schema(self, value: str):
#     def Data(self) -> object:
#     def data(self) -> object:
#     def SetData(self, data: object) -> object:
#     def data(self, value: object):
#     def Extensions(self) -> dict:
#     def extensions(self) -> dict:
#     def SetExtensions(self, extensions: dict) -> object:
#     def extensions(self, value: dict):
#     def ContentType(self) -> str:
#     def content_type(self) -> str:
#     def SetContentType(self, contentType: str) -> object:
#     def content_type(self, value: str):
#     def Properties(self, with_nullable=False) -> dict:
#     def Get(self, key: str) -> (object, bool):
#     def Set(self, key: str, value: object):
#     def MarshalJSON(self, data_marshaller: types.MarshallerType) -> str:
#     def UnmarshalJSON(
#         self,
#         b: typing.Union[str, bytes],
#         data_unmarshaller: types.UnmarshallerType,
#     ):
#     def UnmarshalBinary(
#         self,
#         headers: dict,
#         body: typing.Union[bytes, str],
#         data_unmarshaller: types.UnmarshallerType,
#     ):
#     def MarshalBinary(
#         self, data_marshaller: types.MarshallerType
#     ) -> (dict, bytes):


import typing
from cloudevents.sdk import types
from cloudevents.sdk.converters import base
from cloudevents.sdk.converters.util import has_binary_headers
from cloudevents.sdk.event import base as event_base
#    License for the specific language governing permissions and limitations
#    under the License.




# 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] = {}
    ) -> bool:
        return (
            isinstance(content_type, str)
            and content_type.startswith(self.MIME_TYPE)
            or not has_binary_headers(headers)
        )

    def event_supported(self, event: object) -> bool:
        # structured format supported by both spec 0.1 and 0.2
        return True

    def read(
        self,
        event: event_base.BaseEvent,
        headers: dict,
        body: typing.IO,

# Path: cloudevents/sdk/types.py

# 
# Path: cloudevents/sdk/converters/base.py
# class Converter(object):
#     TYPE = None
#     def read(
#         self,
#         event,
#         headers: dict,
#         body: typing.IO,
#         data_unmarshaller: typing.Callable,
#     ) -> base.BaseEvent:
#     def event_supported(self, event: object) -> bool:
#     def can_read(self, content_type: str) -> bool:
#     def write(
#         self, event: base.BaseEvent, data_marshaller: typing.Callable
#     ) -> (dict, object):
# 
# Path: cloudevents/sdk/converters/util.py
# 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
#     )
# 
# Path: cloudevents/sdk/event/base.py
# class EventGetterSetter(object):  # pragma: no cover
# class BaseEvent(EventGetterSetter):
#     def CloudEventVersion(self) -> str:
#     def specversion(self):
#     def SetCloudEventVersion(self, specversion: str) -> object:
#     def specversion(self, value: str):
#     def EventType(self) -> str:
#     def type(self):
#     def SetEventType(self, eventType: str) -> object:
#     def type(self, value: str):
#     def Source(self) -> str:
#     def source(self):
#     def SetSource(self, source: str) -> object:
#     def source(self, value: str):
#     def EventID(self) -> str:
#     def id(self):
#     def SetEventID(self, eventID: str) -> object:
#     def id(self, value: str):
#     def EventTime(self) -> str:
#     def time(self):
#     def SetEventTime(self, eventTime: str) -> object:
#     def time(self, value: str):
#     def SchemaURL(self) -> str:
#     def schema(self) -> str:
#     def SetSchemaURL(self, schemaURL: str) -> object:
#     def schema(self, value: str):
#     def Data(self) -> object:
#     def data(self) -> object:
#     def SetData(self, data: object) -> object:
#     def data(self, value: object):
#     def Extensions(self) -> dict:
#     def extensions(self) -> dict:
#     def SetExtensions(self, extensions: dict) -> object:
#     def extensions(self, value: dict):
#     def ContentType(self) -> str:
#     def content_type(self) -> str:
#     def SetContentType(self, contentType: str) -> object:
#     def content_type(self, value: str):
#     def Properties(self, with_nullable=False) -> dict:
#     def Get(self, key: str) -> (object, bool):
#     def Set(self, key: str, value: object):
#     def MarshalJSON(self, data_marshaller: types.MarshallerType) -> str:
#     def UnmarshalJSON(
#         self,
#         b: typing.Union[str, bytes],
#         data_unmarshaller: types.UnmarshallerType,
#     ):
#     def UnmarshalBinary(
#         self,
#         headers: dict,
#         body: typing.Union[bytes, str],
#         data_unmarshaller: types.UnmarshallerType,
#     ):
#     def MarshalBinary(
#         self, data_marshaller: types.MarshallerType
#     ) -> (dict, bytes):


# Path: cloudevents/sdk/converters/structured.py
import typing
from cloudevents.sdk import types
from cloudevents.sdk.converters import base
from cloudevents.sdk.converters.util import has_binary_headers
from cloudevents.sdk.event import base as event_base

#    License for the specific language governing permissions and limitations
#    under the License.




# 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] = {}
    ) -> bool:
        return (
            isinstance(content_type, str)
            and content_type.startswith(self.MIME_TYPE)
            or not has_binary_headers(headers)
        )

    def event_supported(self, event: object) -> bool:
        # structured format supported by both spec 0.1 and 0.2
        return True

    def read(
        self,
        event: event_base.BaseEvent,
        headers: dict,
        body: typing.IO,

        data_unmarshaller: types.UnmarshallerType,