CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-cloudevents

CloudEvents Python SDK for creating, sending, and receiving CloudEvents over HTTP in both binary and structured content modes

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

legacy-functions.mddocs/

Legacy HTTP Functions

Deprecated conversion functions maintained for backward compatibility. These functions provide binary and structured format conversion but are superseded by newer APIs in the main HTTP module.

⚠️ Note: These functions are deprecated and maintained only for backward compatibility. New code should use the main cloudevents.http module functions instead.

Capabilities

Binary Format Conversion (Deprecated)

Converts CloudEvent to binary HTTP format where attributes are stored as HTTP headers with ce- prefix and data in the body.

def to_binary(event: CloudEvent, 
              data_marshaller: Optional[MarshallerType] = None) -> Tuple[Dict[str, str], bytes]:
    """
    Converts CloudEvent to binary HTTP format.
    
    DEPRECATED: Use newer HTTP conversion methods instead.
    
    Args:
        event: CloudEvent to convert
        data_marshaller: Optional function to serialize event data
        
    Returns:
        Tuple of (headers dict, body bytes) in binary format
        
    Raises:
        DataMarshallerError: If data marshalling fails
    """

def to_binary_http(event: CloudEvent,
                   data_marshaller: Optional[MarshallerType] = None) -> Tuple[Dict[str, str], bytes]:
    """
    Converts CloudEvent to binary HTTP format.
    
    DEPRECATED: Alias for to_binary function.
    
    Args:
        event: CloudEvent to convert  
        data_marshaller: Optional function to serialize event data
        
    Returns:
        Tuple of (headers dict, body bytes) in binary format
    """

Usage Example

from cloudevents.http import to_binary, CloudEvent

# Create CloudEvent
event = CloudEvent({
    "type": "com.example.string",
    "source": "https://example.com/source"
}, {"key": "value"})

# Convert to binary format (deprecated)
headers, body = to_binary(event)
print(f"Headers: {headers}")  # Contains ce-type, ce-source, etc.
print(f"Body: {body}")        # Contains serialized data

Structured Format Conversion (Deprecated)

Converts CloudEvent to structured HTTP format where the entire event is serialized as JSON in the body.

def to_structured(event: CloudEvent,
                  data_marshaller: Optional[MarshallerType] = None) -> bytes:
    """
    Converts CloudEvent to structured HTTP format.
    
    DEPRECATED: Use newer HTTP conversion methods instead.
    
    Args:
        event: CloudEvent to convert
        data_marshaller: Optional function to serialize event data
        
    Returns:
        Serialized CloudEvent as JSON bytes
        
    Raises:
        DataMarshallerError: If data marshalling fails
    """

def to_structured_http(event: CloudEvent,
                       data_marshaller: Optional[MarshallerType] = None) -> Tuple[Dict[str, str], bytes]:
    """
    Converts CloudEvent to structured HTTP format with headers.
    
    DEPRECATED: Use newer HTTP conversion methods instead.
    
    Args:
        event: CloudEvent to convert
        data_marshaller: Optional function to serialize event data
        
    Returns:
        Tuple of (headers dict, body bytes) with content-type header
    """

Usage Example

from cloudevents.http import to_structured, CloudEvent

# Create CloudEvent
event = CloudEvent({
    "type": "com.example.string", 
    "source": "https://example.com/source",
    "datacontenttype": "application/json"
}, {"key": "value"})

# Convert to structured format (deprecated)
json_bytes = to_structured(event)
print(f"Structured JSON: {json_bytes.decode()}")

JSON Conversion (Deprecated)

Converts CloudEvent to JSON string representation.

def to_json(event: CloudEvent,
            data_marshaller: Optional[MarshallerType] = None) -> str:
    """
    Converts CloudEvent to JSON string.
    
    DEPRECATED: Use newer HTTP conversion methods instead.
    
    Args:
        event: CloudEvent to convert
        data_marshaller: Optional function to serialize event data
        
    Returns:
        CloudEvent as JSON string
        
    Raises:
        DataMarshallerError: If data marshalling fails
    """

Usage Example

from cloudevents.http import to_json, CloudEvent

# Create CloudEvent
event = CloudEvent({
    "type": "com.example.string",
    "source": "https://example.com/source"
}, {"message": "Hello World"})

# Convert to JSON string (deprecated)
json_str = to_json(event)
print(f"JSON representation: {json_str}")

Format Detection (Deprecated)

Functions to detect CloudEvent format from HTTP headers.

def is_binary(headers: Dict[str, str]) -> bool:  
    """
    Checks if headers indicate binary CloudEvent format.
    
    DEPRECATED: Use newer format detection methods instead.
    
    Args:
        headers: HTTP headers dictionary
        
    Returns:
        True if headers indicate binary format, False otherwise
    """

def is_structured(headers: Dict[str, str]) -> bool:
    """
    Checks if headers indicate structured CloudEvent format.
    
    DEPRECATED: Use newer format detection methods instead.
    
    Args:
        headers: HTTP headers dictionary
        
    Returns:
        True if headers indicate structured format, False otherwise
    """

Usage Example

from cloudevents.http import is_binary, is_structured

# Binary format headers
binary_headers = {
    'ce-specversion': '1.0',
    'ce-type': 'com.example.string',
    'ce-source': 'https://example.com/source'
}

print(f"Is binary: {is_binary(binary_headers)}")      # True
print(f"Is structured: {is_structured(binary_headers)}")  # False

# Structured format headers  
structured_headers = {
    'content-type': 'application/cloudevents+json'
}

print(f"Is binary: {is_binary(structured_headers)}")      # False  
print(f"Is structured: {is_structured(structured_headers)}")  # True

Migration Guide

From Legacy to Modern API

Old (Deprecated):

from cloudevents.http import to_binary, to_structured, to_json, is_binary
from cloudevents.http import CloudEvent

event = CloudEvent(attributes, data)
headers, body = to_binary(event)
json_str = to_json(event)

New (Recommended):

from cloudevents.http import CloudEvent, from_json, from_http
from cloudevents.conversion import to_dict

event = CloudEvent(attributes, data) 
# Use modern parsing functions
parsed_event = from_http(headers, body)
# Use conversion utilities
event_dict = to_dict(event)

Why Migrate?

  1. Better Error Handling: Modern functions provide more specific exception types
  2. Improved Type Safety: Better type annotations and IDE support
  3. Consistent API: Unified interface across different transports
  4. Active Maintenance: New features and bug fixes focus on modern API
  5. Performance: Optimized implementations in newer functions

Types

# Type aliases used in legacy functions
MarshallerType = Callable[[Any], AnyStr]
UnmarshallerType = Callable[[AnyStr], Any]

Install with Tessl CLI

npx tessl i tessl/pypi-cloudevents

docs

http-operations.md

index.md

kafka-integration.md

legacy-functions.md

pydantic-validation.md

tile.json