HTTP CloudEvent Operations

class CloudEvent:
    """
    Python-friendly cloudevent class supporting v1 events.
    Supports both binary and structured mode CloudEvents.
    """
    
    def __init__(self, attributes: Mapping[str, str], data: Any = None):
        """
        Event Constructor
        
        Args:
            attributes: A dict with cloudevent attributes. Minimally expects 
                       'type' and 'source'. If not given 'specversion', 'id' 
                       or 'time', creates those with default values.
            data: The payload of the CloudEvent instance. Can be any type.
        """
        
    @classmethod  
    def create(cls, attributes: Mapping[str, Any], data: Optional[Any]) -> CloudEvent:
        """
        Creates a new instance of the CloudEvent using supplied attributes and data.
        
        Args:
            attributes: The attributes of the CloudEvent instance
            data: The payload of the CloudEvent instance
            
        Returns:
            A new CloudEvent instance
        """
        
    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"""
        
    def __delitem__(self, key: str) -> None:
        """Delete a CloudEvent attribute"""
        
    def __contains__(self, key: str) -> bool:
        """Determines if an attribute with a given key is present"""

from cloudevents.http import CloudEvent

# Create event with minimal required attributes
attributes = {
    "type": "com.example.orders.created", 
    "source": "https://example.com/orders"
}

data = {
    "order_id": "12345",
    "customer": "john@example.com",
    "amount": 99.99
}

event = CloudEvent(attributes, data)

# Access properties using dictionary-like interface
print(f"Event ID: {event['id']}")  # Auto-generated UUID
print(f"Event type: {event['type']}")
print(f"Event data: {event.get_data()}")

# Alternative access using get() method
print(f"Event source: {event.get('source')}")
print(f"Event time: {event.get('time')}")

def from_json(data: Union[str, bytes], 
              data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
    """
    Parses JSON string data into a CloudEvent.
    
    Args:
        data: JSON string representation of a CloudEvent
        data_unmarshaller: Optional callable function that casts data to a Python object
        
    Returns:
        A CloudEvent parsed from the given JSON representation
        
    Raises:
        InvalidStructuredJSON: If JSON format is invalid
        MissingRequiredFields: If required fields are missing
    """

from cloudevents.http import from_json
import json

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

# Parse from JSON
event = from_json(json_data)
print(f"Parsed event type: {event['type']}")

def from_http(headers: Union[Mapping[str, str], SupportsDuplicateItems[str, str]], 
              data: Optional[Union[str, bytes]],
              data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
    """
    Parses CloudEvent data and headers into a CloudEvent.
    
    The method supports both binary and structured representations.
    
    Args:
        headers: HTTP headers from the request
        data: HTTP body data (optional for events without payload)
        data_unmarshaller: Optional callable function that casts data to a Python object
        
    Returns:
        A CloudEvent parsed from the HTTP request
        
    Raises:
        InvalidHeadersFormat: If headers format is invalid
        MissingRequiredFields: If required CloudEvent fields are missing
    """

from cloudevents.http import from_http

# Binary mode example
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"}'

event = from_http(headers, data)

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

event = from_http(headers, data)

def from_dict(data: Mapping[str, Any], 
              data_unmarshaller: Optional[UnmarshallerType] = None) -> CloudEvent:
    """
    Creates CloudEvent from dictionary representation.
    
    Args:
        data: Dictionary containing CloudEvent attributes and data
        data_unmarshaller: Optional callable function that casts data to a Python object
        
    Returns:
        A CloudEvent created from the dictionary
        
    Raises:
        MissingRequiredFields: If required fields are missing
        InvalidRequiredFields: If required fields are invalid
    """

from cloudevents.http import from_dict

# Dictionary representation  
event_dict = {
    "specversion": "1.0",
    "type": "com.example.orders.created",
    "source": "https://example.com/orders",
    "id": "order-123",
    "datacontenttype": "application/json", 
    "data": {
        "order_id": "12345",
        "amount": 99.99
    }
}

event = from_dict(event_dict)
print(f"Created event: {event['id']}")

# Type aliases used in HTTP operations
MarshallerType = Callable[[Any], AnyStr]
UnmarshallerType = Callable[[AnyStr], Any]

# Protocol for headers that may have duplicate items (like HTTP headers)
class SupportsDuplicateItems(Protocol[_K_co, _V_co]):
    def items(self) -> Iterable[Tuple[_K_co, _V_co]]: ...