0
# HTTP CloudEvent Operations
1
2
Core CloudEvent creation, parsing, and conversion functionality for HTTP transport. Supports both binary and structured content modes with automatic format detection and comprehensive attribute management.
3
4
## Capabilities
5
6
### CloudEvent Class
7
8
The main CloudEvent class for HTTP transport, supporting both binary and structured modes with full CloudEvents v1.0 specification compliance.
9
10
```python { .api }
11
class CloudEvent:
12
"""
13
Python-friendly cloudevent class supporting v1 events.
14
Supports both binary and structured mode CloudEvents.
15
"""
16
17
def __init__(self, attributes: Mapping[str, str], data: Any = None):
18
"""
19
Event Constructor
20
21
Args:
22
attributes: A dict with cloudevent attributes. Minimally expects
23
'type' and 'source'. If not given 'specversion', 'id'
24
or 'time', creates those with default values.
25
data: The payload of the CloudEvent instance. Can be any type.
26
"""
27
28
@classmethod
29
def create(cls, attributes: Mapping[str, Any], data: Optional[Any]) -> CloudEvent:
30
"""
31
Creates a new instance of the CloudEvent using supplied attributes and data.
32
33
Args:
34
attributes: The attributes of the CloudEvent instance
35
data: The payload of the CloudEvent instance
36
37
Returns:
38
A new CloudEvent instance
39
"""
40
41
def get_data(self) -> Optional[Any]:
42
"""Get the event data payload"""
43
44
def get_attributes(self) -> Mapping[str, Any]:
45
"""Returns a read-only view on the attributes of the event"""
46
47
def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
48
"""
49
Retrieves an event attribute value for the given key.
50
Returns the default value if the attribute does not exist.
51
"""
52
53
def __getitem__(self, key: str) -> Any:
54
"""Returns a value of an attribute of the event denoted by the given key"""
55
56
def __setitem__(self, key: str, value: Any) -> None:
57
"""Set a CloudEvent attribute"""
58
59
def __delitem__(self, key: str) -> None:
60
"""Delete a CloudEvent attribute"""
61
62
def __contains__(self, key: str) -> bool:
63
"""Determines if an attribute with a given key is present"""
64
```
65
66
#### Usage Example
67
68
```python
69
from cloudevents.http import CloudEvent
70
71
# Create event with minimal required attributes
72
attributes = {
73
"type": "com.example.orders.created",
74
"source": "https://example.com/orders"
75
}
76
77
data = {
78
"order_id": "12345",
79
"customer": "john@example.com",
80
"amount": 99.99
81
}
82
83
event = CloudEvent(attributes, data)
84
85
# Access properties using dictionary-like interface
86
print(f"Event ID: {event['id']}") # Auto-generated UUID
87
print(f"Event type: {event['type']}")
88
print(f"Event data: {event.get_data()}")
89
90
# Alternative access using get() method
91
print(f"Event source: {event.get('source')}")
92
print(f"Event time: {event.get('time')}")
93
```
94
95
### JSON Parsing
96
97
Parses JSON string representations of CloudEvents into CloudEvent objects, with support for custom data unmarshalling.
98
99
```python { .api }
100
def from_json(data: Union[str, bytes],
101
data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
102
"""
103
Parses JSON string data into a CloudEvent.
104
105
Args:
106
data: JSON string representation of a CloudEvent
107
data_unmarshaller: Optional callable function that casts data to a Python object
108
109
Returns:
110
A CloudEvent parsed from the given JSON representation
111
112
Raises:
113
InvalidStructuredJSON: If JSON format is invalid
114
MissingRequiredFields: If required fields are missing
115
"""
116
```
117
118
#### Usage Example
119
120
```python
121
from cloudevents.http import from_json
122
import json
123
124
# JSON representation of a CloudEvent
125
json_data = json.dumps({
126
"specversion": "1.0",
127
"type": "com.example.string",
128
"source": "https://example.com/source",
129
"id": "A234-1234-1234",
130
"datacontenttype": "application/json",
131
"data": {"key": "value"}
132
})
133
134
# Parse from JSON
135
event = from_json(json_data)
136
print(f"Parsed event type: {event['type']}")
137
```
138
139
### HTTP Parsing
140
141
Parses CloudEvent data and headers from HTTP requests, supporting both binary and structured content modes with automatic format detection.
142
143
```python { .api }
144
def from_http(headers: Union[Mapping[str, str], SupportsDuplicateItems[str, str]],
145
data: Optional[Union[str, bytes]],
146
data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
147
"""
148
Parses CloudEvent data and headers into a CloudEvent.
149
150
The method supports both binary and structured representations.
151
152
Args:
153
headers: HTTP headers from the request
154
data: HTTP body data (optional for events without payload)
155
data_unmarshaller: Optional callable function that casts data to a Python object
156
157
Returns:
158
A CloudEvent parsed from the HTTP request
159
160
Raises:
161
InvalidHeadersFormat: If headers format is invalid
162
MissingRequiredFields: If required CloudEvent fields are missing
163
"""
164
```
165
166
#### Usage Example
167
168
```python
169
from cloudevents.http import from_http
170
171
# Binary mode example
172
headers = {
173
'ce-specversion': '1.0',
174
'ce-type': 'com.example.string',
175
'ce-source': 'https://example.com/source',
176
'ce-id': 'A234-1234-1234',
177
'content-type': 'application/json'
178
}
179
data = b'{"key": "value"}'
180
181
event = from_http(headers, data)
182
183
# Structured mode example
184
headers = {'content-type': 'application/cloudevents+json'}
185
data = b'{"specversion": "1.0", "type": "com.example.string", "source": "https://example.com/source", "id": "A234-1234-1234", "data": {"key": "value"}}'
186
187
event = from_http(headers, data)
188
```
189
190
### Dictionary Parsing
191
192
Creates CloudEvent objects from dictionary representations, useful for programmatic event creation and serialization.
193
194
```python { .api }
195
def from_dict(data: Mapping[str, Any],
196
data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
197
"""
198
Creates CloudEvent from dictionary representation.
199
200
Args:
201
data: Dictionary containing CloudEvent attributes and data
202
data_unmarshaller: Optional callable function that casts data to a Python object
203
204
Returns:
205
A CloudEvent created from the dictionary
206
207
Raises:
208
MissingRequiredFields: If required fields are missing
209
InvalidRequiredFields: If required fields are invalid
210
"""
211
```
212
213
#### Usage Example
214
215
```python
216
from cloudevents.http import from_dict
217
218
# Dictionary representation
219
event_dict = {
220
"specversion": "1.0",
221
"type": "com.example.orders.created",
222
"source": "https://example.com/orders",
223
"id": "order-123",
224
"datacontenttype": "application/json",
225
"data": {
226
"order_id": "12345",
227
"amount": 99.99
228
}
229
}
230
231
event = from_dict(event_dict)
232
print(f"Created event: {event['id']}")
233
```
234
235
## Types
236
237
```python { .api }
238
# Type aliases used in HTTP operations
239
MarshallerType = Callable[[Any], AnyStr]
240
UnmarshallerType = Callable[[AnyStr], Any]
241
242
# Protocol for headers that may have duplicate items (like HTTP headers)
243
class SupportsDuplicateItems(Protocol[_K_co, _V_co]):
244
def items(self) -> Iterable[Tuple[_K_co, _V_co]]: ...
245
```