Pydantic Validation

pip install cloudevents[pydantic]

class CloudEvent(BaseModel):
    """
    Pydantic-based CloudEvent implementation with validation.
    
    Automatically validates CloudEvent attributes according to the specification
    and provides type safety with IDE support.
    """
    
    specversion: str = Field(..., description="CloudEvents specification version")
    type: str = Field(..., description="Event type identifier")  
    source: str = Field(..., description="Event source identifier")
    id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Event ID")
    time: Optional[datetime] = Field(None, description="Event timestamp")
    datacontenttype: Optional[str] = Field(None, description="Data content type")
    dataschema: Optional[str] = Field(None, description="Data schema URI")
    subject: Optional[str] = Field(None, description="Event subject")
    data: Optional[Any] = Field(None, description="Event data payload")
    
    @classmethod
    def create(cls, attributes: Mapping[str, Any], data: Optional[Any]) -> CloudEvent:
        """
        Creates a new Pydantic CloudEvent instance.
        
        Args:
            attributes: CloudEvent attributes dictionary
            data: Event data payload
            
        Returns:
            Validated CloudEvent instance
            
        Raises:
            ValidationError: If attributes don't meet CloudEvents specification
        """
        
    def get_data(self) -> Optional[Any]:
        """Get the event data payload"""
        
    def get_attributes(self) -> Mapping[str, Any]:
        """Returns a read-only view on the attributes of the event"""
        
    def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
        """
        Retrieves an event attribute value for the given key.
        Returns the default value if the attribute does not exist.
        """
        
    def __getitem__(self, key: str) -> Any:
        """Returns a value of an attribute of the event denoted by the given key"""
        
    def __setitem__(self, key: str, value: Any) -> None:
        """Set a CloudEvent attribute (not for data field)"""
        
    # Pydantic field properties for direct access
    data: Optional[Any]  # Event data payload
    source: str  # Event source (required)
    id: str  # Event ID (auto-generated if not provided)
    type: str  # Event type (required)
    specversion: str  # CloudEvents specification version
    time: Optional[datetime]  # Event timestamp
    datacontenttype: Optional[str]  # Data content type
    dataschema: Optional[str]  # Data schema URI
    subject: Optional[str]  # Event subject

from cloudevents.pydantic import CloudEvent
from datetime import datetime
from pydantic import ValidationError

# Create validated CloudEvent
try:
    event = CloudEvent(
        specversion="1.0",
        type="com.example.orders.created",
        source="https://example.com/orders",
        id="order-123",
        time=datetime.now(),
        datacontenttype="application/json",
        data={"order_id": "12345", "amount": 99.99}
    )
    print(f"Created valid event: {event.id}")
except ValidationError as e:
    print(f"Validation failed: {e}")

# Automatic validation of invalid data
try:
    invalid_event = CloudEvent(
        specversion="2.0",  # Invalid version
        type="",           # Empty type not allowed
        source="not-a-uri" # Invalid URI format  
    )
except ValidationError as e:
    print(f"Validation errors: {e}")

def from_json(data: Union[str, bytes],
              data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
    """
    Parses JSON string into a validated Pydantic CloudEvent.
    
    Args:
        data: JSON string representation of a CloudEvent
        data_unmarshaller: Optional callable function that casts data to a Python object
        
    Returns:
        Validated Pydantic CloudEvent instance
        
    Raises:
        ValidationError: If CloudEvent doesn't meet specification requirements
        InvalidStructuredJSON: If JSON format is invalid
    """

from cloudevents.pydantic import from_json
from pydantic import ValidationError
import json

# Valid JSON
json_data = json.dumps({
    "specversion": "1.0",
    "type": "com.example.string",
    "source": "https://example.com/source",
    "id": "A234-1234-1234", 
    "data": {"key": "value"}
})

try:
    event = from_json(json_data)
    print(f"Parsed validated event: {event.type}")
except ValidationError as e:
    print(f"Validation failed: {e}")

def from_http(headers: Union[Mapping[str, str], SupportsDuplicateItems[str, str]],
              data: Optional[Union[str, bytes]], 
              data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
    """
    Parses HTTP request into a validated Pydantic CloudEvent.
    
    Supports both binary and structured content modes with automatic validation.
    
    Args:
        headers: HTTP headers from the request
        data: HTTP body data
        data_unmarshaller: Optional callable function that casts data to a Python object
        
    Returns:
        Validated Pydantic CloudEvent instance
        
    Raises:
        ValidationError: If CloudEvent doesn't meet specification requirements
        InvalidHeadersFormat: If headers format is invalid
    """

from cloudevents.pydantic import from_http
from pydantic import ValidationError

# Binary mode with validation
headers = {
    'ce-specversion': '1.0',
    'ce-type': 'com.example.string',
    'ce-source': 'https://example.com/source', 
    'ce-id': 'A234-1234-1234',
    'content-type': 'application/json'
}
data = b'{"key": "value"}'

try:
    event = from_http(headers, data)
    print(f"Validated event from HTTP: {event.type}")
except ValidationError as e:
    print(f"Validation failed: {e}")

def from_dict(data: Mapping[str, Any],
              data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
    """
    Creates validated Pydantic CloudEvent from dictionary.
    
    Args:
        data: Dictionary containing CloudEvent attributes and data
        data_unmarshaller: Optional callable function that casts data to a Python object
        
    Returns:
        Validated Pydantic CloudEvent instance
        
    Raises:
        ValidationError: If CloudEvent doesn't meet specification requirements
    """

from cloudevents.pydantic import from_dict
from pydantic import ValidationError

event_dict = {
    "specversion": "1.0",
    "type": "com.example.orders.created",
    "source": "https://example.com/orders",
    "id": "order-123",
    "data": {"order_id": "12345"}
}

try:
    event = from_dict(event_dict)
    print(f"Validated event from dict: {event.id}")
except ValidationError as e:
    print(f"Validation failed: {e}")

from cloudevents.pydantic import CloudEvent
from datetime import datetime

# IDE autocomplete and type checking
event = CloudEvent(
    specversion="1.0",
    type="com.example.event",
    source="https://example.com/source"
)

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

from pydantic import ValidationError

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

# Validates field formats  
try:
    CloudEvent(
        specversion="1.0",
        type="com.example.event", 
        source="not-a-valid-uri",  # Invalid URI format
        time="not-a-datetime"      # Invalid datetime format
    )
except ValidationError as e:
    print("Format validation failed")

from cloudevents.pydantic import CloudEvent
from pydantic import ValidationError

try:
    CloudEvent(
        specversion="2.0",  # Invalid version
        type="",           # Empty type
        source="invalid"   # Invalid URI
    )
except ValidationError as e:
    # Detailed validation error information
    for error in e.errors():
        print(f"Field: {error['loc']}")
        print(f"Error: {error['msg']}")
        print(f"Input: {error['input']}")

# Type aliases used in Pydantic validation
MarshallerType = Callable[[Any], AnyStr]
UnmarshallerType = Callable[[AnyStr], Any]

# Exception raised when Pydantic is not installed
class PydanticFeatureNotInstalled(GenericException):
    """Raised when Pydantic feature is used but not installed"""