cloudevents_pydantic/formats/json.py
# ==============================================================================
# Copyright (c) 2024 Federico Busetti =
# <729029+febus982@users.noreply.github.com> =
# =
# Permission is hereby granted, free of charge, to any person obtaining a =
# copy of this software and associated documentation files (the "Software"), =
# to deal in the Software without restriction, including without limitation =
# the rights to use, copy, modify, merge, publish, distribute, sublicense, =
# and/or sell copies of the Software, and to permit persons to whom the =
# Software is furnished to do so, subject to the following conditions: =
# =
# The above copyright notice and this permission notice shall be included in =
# all copies or substantial portions of the Software. =
# =
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR =
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, =
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL =
# THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER =
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING =
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER =
# DEALINGS IN THE SOFTWARE. =
# ==============================================================================
from typing import List, TypeVar
from pydantic import TypeAdapter
from ..events import CloudEvent
_T = TypeVar("_T", bound=CloudEvent)
def to_json(event: CloudEvent) -> str:
"""
Serializes an event in JSON format.
:param event: The event object to serialize
:type event: CloudEvent
:return: The headers and the body representation of the event
:rtype: str
"""
return event.model_dump_json()
def from_json(
data: str, event_adapter: TypeAdapter[_T] = TypeAdapter(CloudEvent)
) -> _T:
"""
Deserializes an event from JSON format.
:param data: the JSON representation of the event
:type data: str
:param event_adapter: The event class to build
:type event_adapter: Type[CloudEvent]
:return: The deserialized event
:rtype: CloudEvent
"""
return event_adapter.validate_json(data)
def to_json_batch(
events: List[_T],
batch_adapter: TypeAdapter[List[_T]] = TypeAdapter(List[CloudEvent]),
) -> str:
"""
Serializes a list of events in JSON batch format.
:param events: The event object to serialize
:type events: List[CloudEvent]
:param batch_adapter: The pydantic TypeAdapter to use
:type: TypeAdapter[List[CloudEvent]]
:return: The serialized event batch
:rtype: str
"""
return batch_adapter.dump_json(events).decode()
def from_json_batch(
data: str,
batch_adapter: TypeAdapter[List[_T]] = TypeAdapter(List[CloudEvent]),
) -> List[_T]:
"""
Deserializes a list of events from JSON batch format.
:param data: The JSON representation of the event batch
:type data: str
:param batch_adapter: The pydantic TypeAdapter to use
:type: TypeAdapter[List[CloudEvent]]
:return: The deserialized event batch
:rtype: List[CloudEvent]
"""
return batch_adapter.validate_json(data)