feat: implementation for JSON format. #245
Conversation
…tion included. Signed-off-by: Tudor Plugaru <[email protected]>
|
@xSAVIKx This is a draft implementation for JSON format specs. Can you please have a look and provide some feedback about the direction I choose? I have to admit that it's inspired by the Java implementation a bit. |
| class Format(Protocol): | ||
| def read(self, data: Union[str, bytes]) -> BaseCloudEvent: ... | ||
|
|
||
| def write(self, event: BaseCloudEvent) -> str: ... |
There was a problem hiding this comment.
not all formats may be represented as string. it's safer to have bytes here. Same applies to read method data attribute.
| :return: The CloudEvent as a JSON formatted byte array. | ||
| """ | ||
| event_data = event.get_data() | ||
| event_dict: dict[str, Any] = {**event.get_attributes()} |
There was a problem hiding this comment.
why not?
| event_dict: dict[str, Any] = {**event.get_attributes()} | |
| event_dict: dict[str, Any] = dict(event.get_attributes()) |
| return super().default(obj) | ||
|
|
||
|
|
||
| class JSONFormat(Format): |
There was a problem hiding this comment.
how do you envision usage of this class by the end users?
There was a problem hiding this comment.
Technically, there should be no need for this class to be used often, unless someone knows what they are doing.
Since for each binding, we would have to implement both structured and binary modes, this format class would be useful when using structured mode, since the entire CloudEvent class is serialized into JSON/Avro and send as a single message. This format would be injected into that "structured mode CE writer" implementation. Something like
to_structured(event, format):
...
from_structured(data, format):
...I can't see this class being used for binary mode tho, correct me if I'm wrong please. But given that attributes are sent as message metadata, it can only be JSON format by default, no?
Anyway, I think it's worth having a way for having CE -> bytes and bytes -> CE for each supported format as per the specs and let engineers do manipulations as they see fit.
Can you be more explicit, please? Batch mode is clear, but what does "binary" mean here? |
| class BaseCloudEvent(Protocol): | ||
| def get_id(self) -> str: ... | ||
|
|
||
| def get_source(self) -> str: ... | ||
|
|
||
| def get_type(self) -> str: ... | ||
|
|
||
| def get_specversion(self) -> str: ... | ||
|
|
||
| def get_datacontenttype(self) -> Optional[str]: ... | ||
|
|
||
| def get_dataschema(self) -> Optional[str]: ... | ||
|
|
||
| def get_subject(self) -> Optional[str]: ... | ||
|
|
||
| def get_time(self) -> Optional[datetime]: ... | ||
|
|
||
| def get_extension(self, extension_name: str) -> Any: ... | ||
|
|
||
| def get_data(self) -> Optional[Union[dict, str, bytes]]: ... | ||
|
|
||
| def get_attributes(self) -> dict[str, Any]: ... |
There was a problem hiding this comment.
It's probably a good idea to move the docs for these methods from the implementation to protocol.
| class Format(Protocol): | ||
| def read(self, data: Union[str, bytes]) -> BaseCloudEvent: ... | ||
|
|
||
| def write(self, event: BaseCloudEvent) -> str: ... |
There was a problem hiding this comment.
So in order to make this useful as a generic approach we should add generics :-)
| class Format(Protocol): | |
| def read(self, data: Union[str, bytes]) -> BaseCloudEvent: ... | |
| def write(self, event: BaseCloudEvent) -> str: ... | |
| T = TypeVar("T", bound=BaseCloudEvent) | |
| class Format(Protocol[T]): | |
| def read(self, data: bytes) -> T: ... | |
| def write(self, event: T) -> bytes: ... |
this way we should be able to get parameterized results as well
There was a problem hiding this comment.
good idea, yes, thanks, I'll do that
Changes
One line description for the changelog