cloudevents_json_serialization_and_deserialization_quickstart.py

1from cloudevents.http import CloudEvent
2from cloudevents.conversion import to_json, from_json
3
4# Create a CloudEvent
5# The event defines the minimum required attributes: id, source, type
6attributes = {
7    "type": "com.example.sampletype1",
8    "source": "https://example.com/event-source",
9}
10data = {"message": "Hello World!"}
11event = CloudEvent(attributes, data)
12
13# Serialize the CloudEvent to JSON (Binary or Structured mode)
14# Here we demonstrate structured encoding
15event_json = to_json(event)
16
17print(f"Serialized event: {event_json.decode('utf-8')}")
18
19# Deserialize the JSON back into a CloudEvent object
20replayed_event = from_json(event_json)
21
22print(f"Deserialized Data: {replayed_event.data['message']}")
23print(f"Event ID: {replayed_event['id']}")