CloudEvents Python SDK

Warnings

• breaking The CloudEvent attribute `contenttype` was renamed to `datacontenttype` in version 2.0.0 to align with the CloudEvents specification. Code using `event.contenttype` will break.
  Fix: Update all instances of `contenttype` to `datacontenttype` when creating or accessing CloudEvent attributes (e.g., `event.attributes['datacontenttype']`).
• breaking In version 2.0.0+, `CloudEvent.data` now consistently returns `bytes` or `None`. In earlier versions (1.x), it could return the original Python object type (e.g., `dict`, `str`).
  Fix: If you expect JSON or text data, you must decode `event.data`. For example, `json.loads(event.data.decode('utf-8'))` for JSON, or `event.data.decode('utf-8')` for plain text.
• breaking Several utility functions and methods, such as `CloudEvent.FromInstance`, `CloudEvent.ToJsonString`, and direct `json.dumps` methods on CloudEvent objects, were removed in version 2.0.0.
  Fix: Use the provided binding functions like `from_structured`, `to_structured` for HTTP operations, or manually serialize/deserialize using `event.attributes` and `event.data`.
• gotcha The `time` attribute of a CloudEvent must conform to RFC 3339 (e.g., `2018-04-05T17:31:00Z`). Incorrectly formatted timestamps might be accepted by the SDK but could cause issues with downstream consumers or spec validation.
  Fix: Ensure `time` is a string formatted as `YYYY-MM-DDTHH:MM:SSZ` or with milliseconds (`YYYY-MM-DDTHH:MM:SS.ffffffZ`). Use `datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='milliseconds').replace('+00:00', 'Z')` for reliable generation.

Install

• pip install cloudevents Install core library

Imports

• CloudEvent

  from cloudevents.http import CloudEvent

  Pre-2.0 versions used versioned imports like `cloudevents.sdk.event.v1`. V2.0.0 consolidates to `cloudevents.http` for HTTP binding.
• to_structured

  from cloudevents.http import to_structured

• from_structured

  from cloudevents.http import from_structured

Quickstart

This example demonstrates how to create a CloudEvent with specified attributes and JSON data, then serialize it into HTTP structured mode headers and body. It also shows how to access the event's data (which is bytes in v2.0.0+).

from cloudevents.http import CloudEvent, to_structured
import json
import datetime

# Define CloudEvent attributes
attributes = {
    "type": "com.example.someevent",
    "source": "/mycontext/myapp",
    "id": "A234-1234-1234",
    "time": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='milliseconds').replace('+00:00', 'Z'),
    "datacontenttype": "application/json"
}

# Define event data
data = {"message": "Hello, CloudEvents!", "value": 123}

# Create a CloudEvent instance
event = CloudEvent(attributes, data)

# Serialize the event to structured HTTP headers and body
headers, body = to_structured(event)

print("--- CloudEvent (Structured Mode) ---")
print("Headers:", headers)
print("Body:", body.decode('utf-8'))
print("\nEvent Data (original):")
print(event.data) # For v2.0.0, this returns bytes or None
print("Event Data (decoded):", json.loads(event.data.decode('utf-8')))