0
# Pydantic Validation
1

2
Optional Pydantic-based CloudEvent implementation providing runtime validation, type safety, and IDE support with automatic serialization/deserialization. Requires the `pydantic` extra to be installed.
3

4
## Installation
5

6
```bash
7
pip install cloudevents[pydantic]
8
```
9

10
## Capabilities
11

12
### Pydantic CloudEvent Class
13

14
Pydantic-based CloudEvent implementation with automatic validation, type conversion, and enhanced developer experience through IDE support.
15

16
```python { .api }
17
class CloudEvent(BaseModel):
18
    """
19
    Pydantic-based CloudEvent implementation with validation.
20
    
21
    Automatically validates CloudEvent attributes according to the specification
22
    and provides type safety with IDE support.
23
    """
24
    
25
    specversion: str = Field(..., description="CloudEvents specification version")
26
    type: str = Field(..., description="Event type identifier")  
27
    source: str = Field(..., description="Event source identifier")
28
    id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Event ID")
29
    time: Optional[datetime] = Field(None, description="Event timestamp")
30
    datacontenttype: Optional[str] = Field(None, description="Data content type")
31
    dataschema: Optional[str] = Field(None, description="Data schema URI")
32
    subject: Optional[str] = Field(None, description="Event subject")
33
    data: Optional[Any] = Field(None, description="Event data payload")
34
    
35
    @classmethod
36
    def create(cls, attributes: Mapping[str, Any], data: Optional[Any]) -> CloudEvent:
37
        """
38
        Creates a new Pydantic CloudEvent instance.
39
        
40
        Args:
41
            attributes: CloudEvent attributes dictionary
42
            data: Event data payload
43
            
44
        Returns:
45
            Validated CloudEvent instance
46
            
47
        Raises:
48
            ValidationError: If attributes don't meet CloudEvents specification
49
        """
50
        
51
    def get_data(self) -> Optional[Any]:
52
        """Get the event data payload"""
53
        
54
    def get_attributes(self) -> Mapping[str, Any]:
55
        """Returns a read-only view on the attributes of the event"""
56
        
57
    def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
58
        """
59
        Retrieves an event attribute value for the given key.
60
        Returns the default value if the attribute does not exist.
61
        """
62
        
63
    def __getitem__(self, key: str) -> Any:
64
        """Returns a value of an attribute of the event denoted by the given key"""
65
        
66
    def __setitem__(self, key: str, value: Any) -> None:
67
        """Set a CloudEvent attribute (not for data field)"""
68
        
69
    # Pydantic field properties for direct access
70
    data: Optional[Any]  # Event data payload
71
    source: str  # Event source (required)
72
    id: str  # Event ID (auto-generated if not provided)
73
    type: str  # Event type (required)
74
    specversion: str  # CloudEvents specification version
75
    time: Optional[datetime]  # Event timestamp
76
    datacontenttype: Optional[str]  # Data content type
77
    dataschema: Optional[str]  # Data schema URI
78
    subject: Optional[str]  # Event subject
79
```
80

81
#### Usage Example
82

83
```python
84
from cloudevents.pydantic import CloudEvent
85
from datetime import datetime
86
from pydantic import ValidationError
87

88
# Create validated CloudEvent
89
try:
90
    event = CloudEvent(
91
        specversion="1.0",
92
        type="com.example.orders.created",
93
        source="https://example.com/orders",
94
        id="order-123",
95
        time=datetime.now(),
96
        datacontenttype="application/json",
97
        data={"order_id": "12345", "amount": 99.99}
98
    )
99
    print(f"Created valid event: {event.id}")
100
except ValidationError as e:
101
    print(f"Validation failed: {e}")
102

103
# Automatic validation of invalid data
104
try:
105
    invalid_event = CloudEvent(
106
        specversion="2.0",  # Invalid version
107
        type="",           # Empty type not allowed
108
        source="not-a-uri" # Invalid URI format  
109
    )
110
except ValidationError as e:
111
    print(f"Validation errors: {e}")
112
```
113

114
### JSON Parsing with Validation
115

116
Parses JSON string representations into validated Pydantic CloudEvent objects with comprehensive error reporting.
117

118
```python { .api }
119
def from_json(data: Union[str, bytes],
120
              data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
121
    """
122
    Parses JSON string into a validated Pydantic CloudEvent.
123
    
124
    Args:
125
        data: JSON string representation of a CloudEvent
126
        data_unmarshaller: Optional callable function that casts data to a Python object
127
        
128
    Returns:
129
        Validated Pydantic CloudEvent instance
130
        
131
    Raises:
132
        ValidationError: If CloudEvent doesn't meet specification requirements
133
        InvalidStructuredJSON: If JSON format is invalid
134
    """
135
```
136

137
#### Usage Example
138

139
```python
140
from cloudevents.pydantic import from_json
141
from pydantic import ValidationError
142
import json
143

144
# Valid JSON
145
json_data = json.dumps({
146
    "specversion": "1.0",
147
    "type": "com.example.string",
148
    "source": "https://example.com/source",
149
    "id": "A234-1234-1234", 
150
    "data": {"key": "value"}
151
})
152

153
try:
154
    event = from_json(json_data)
155
    print(f"Parsed validated event: {event.type}")
156
except ValidationError as e:
157
    print(f"Validation failed: {e}")
158
```
159

160
### HTTP Parsing with Validation
161

162
Parses HTTP headers and data into validated Pydantic CloudEvent objects with automatic format detection and validation.
163

164
```python { .api }
165
def from_http(headers: Union[Mapping[str, str], SupportsDuplicateItems[str, str]],
166
              data: Optional[Union[str, bytes]], 
167
              data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
168
    """
169
    Parses HTTP request into a validated Pydantic CloudEvent.
170
    
171
    Supports both binary and structured content modes with automatic validation.
172
    
173
    Args:
174
        headers: HTTP headers from the request
175
        data: HTTP body data
176
        data_unmarshaller: Optional callable function that casts data to a Python object
177
        
178
    Returns:
179
        Validated Pydantic CloudEvent instance
180
        
181
    Raises:
182
        ValidationError: If CloudEvent doesn't meet specification requirements
183
        InvalidHeadersFormat: If headers format is invalid
184
    """
185
```
186

187
#### Usage Example
188

189
```python
190
from cloudevents.pydantic import from_http
191
from pydantic import ValidationError
192

193
# Binary mode with validation
194
headers = {
195
    'ce-specversion': '1.0',
196
    'ce-type': 'com.example.string',
197
    'ce-source': 'https://example.com/source', 
198
    'ce-id': 'A234-1234-1234',
199
    'content-type': 'application/json'
200
}
201
data = b'{"key": "value"}'
202

203
try:
204
    event = from_http(headers, data)
205
    print(f"Validated event from HTTP: {event.type}")
206
except ValidationError as e:
207
    print(f"Validation failed: {e}")
208
```
209

210
### Dictionary Parsing with Validation
211

212
Creates validated Pydantic CloudEvent objects from dictionary representations with comprehensive validation.
213

214
```python { .api }
215
def from_dict(data: Mapping[str, Any],
216
              data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
217
    """
218
    Creates validated Pydantic CloudEvent from dictionary.
219
    
220
    Args:
221
        data: Dictionary containing CloudEvent attributes and data
222
        data_unmarshaller: Optional callable function that casts data to a Python object
223
        
224
    Returns:
225
        Validated Pydantic CloudEvent instance
226
        
227
    Raises:
228
        ValidationError: If CloudEvent doesn't meet specification requirements
229
    """
230
```
231

232
#### Usage Example
233

234
```python
235
from cloudevents.pydantic import from_dict
236
from pydantic import ValidationError
237

238
event_dict = {
239
    "specversion": "1.0",
240
    "type": "com.example.orders.created",
241
    "source": "https://example.com/orders",
242
    "id": "order-123",
243
    "data": {"order_id": "12345"}
244
}
245

246
try:
247
    event = from_dict(event_dict)
248
    print(f"Validated event from dict: {event.id}")
249
except ValidationError as e:
250
    print(f"Validation failed: {e}")
251
```
252

253
## Validation Benefits
254

255
### Type Safety
256

257
```python
258
from cloudevents.pydantic import CloudEvent
259
from datetime import datetime
260

261
# IDE autocomplete and type checking
262
event = CloudEvent(
263
    specversion="1.0",
264
    type="com.example.event",
265
    source="https://example.com/source"
266
)
267

268
# Type-safe access with IDE support
269
event_time: datetime = event.time  # Optional[datetime] 
270
event_type: str = event.type       # str (required)
271
```
272

273
### Automatic Validation
274

275
```python
276
from pydantic import ValidationError
277

278
# Validates required fields
279
try:
280
    CloudEvent(specversion="1.0")  # Missing required 'type' and 'source'
281
except ValidationError as e:
282
    print("Missing required fields detected")
283

284
# Validates field formats  
285
try:
286
    CloudEvent(
287
        specversion="1.0",
288
        type="com.example.event", 
289
        source="not-a-valid-uri",  # Invalid URI format
290
        time="not-a-datetime"      # Invalid datetime format
291
    )
292
except ValidationError as e:
293
    print("Format validation failed")
294
```
295

296
### Enhanced Error Messages
297

298
```python
299
from cloudevents.pydantic import CloudEvent
300
from pydantic import ValidationError
301

302
try:
303
    CloudEvent(
304
        specversion="2.0",  # Invalid version
305
        type="",           # Empty type
306
        source="invalid"   # Invalid URI
307
    )
308
except ValidationError as e:
309
    # Detailed validation error information
310
    for error in e.errors():
311
        print(f"Field: {error['loc']}")
312
        print(f"Error: {error['msg']}")
313
        print(f"Input: {error['input']}")
314
```
315

316
## Version Compatibility
317

318
The Pydantic integration automatically detects and uses the appropriate Pydantic version:
319

320
- **Pydantic v1**: Uses `cloudevents.pydantic.v1` implementation
321
- **Pydantic v2**: Uses `cloudevents.pydantic.v2` implementation  
322

323
Both versions provide the same API interface for seamless compatibility.
324

325
## Types
326

327
```python { .api }
328
# Type aliases used in Pydantic validation
329
MarshallerType = Callable[[Any], AnyStr]
330
UnmarshallerType = Callable[[AnyStr], Any]
331

332
# Exception raised when Pydantic is not installed
333
class PydanticFeatureNotInstalled(GenericException):
334
    """Raised when Pydantic feature is used but not installed"""
335
```