cloudevents-pydantic 0.2.0

An implementation of the CloudEvents spec using Pydantic V2

cloudevents-pydantic

This is an implementation of the CloudEvents spec using Pydantic V2 for high performance during validation and serialization.

It is meant to support natively FastAPI and FastStream (WIP)

How to use

pip install cloudevents-pydantic

from cloudevents_pydantic.bindings.http import HTTPHandler
from cloudevents_pydantic.events import CloudEvent

handler = HTTPHandler()
minimal_attributes = {
    "type": "com.example.string",
    "source": "https://example.com/event-producer",
}

# `CloudEvent` is a Pydantic model to handle validation and serialization
# `event_factory` is a helper method to autogenerate some of the mandatory 
# such as id, time, specversion
event = CloudEvent.event_factory(**minimal_attributes)

# Single event HTTP serialization
headers, single_event_as_json = handler.to_json(event)

# Batch of events HTTP serialization
headers, batch_of_events_as_json = handler.to_json_batch([event])

# Parsing a JSON string for a single event
parsed_event = handler.from_json(single_event_as_json)

# Parsing a JSON string for a single event
parsed_event_list = handler.from_json(batch_of_events_as_json)

Refer to the docs for more advanced use cases and for details on how to create custom events.

Performance

Using pydantic gives a great performance boost if compared to the official SDK. (there's obviously some performance issue in the official serialization using pydantic)

These results come from a Macbook Pro M4 Pro on python 3.13. Feel free to run the benchmark.py script yourself.

==== 1M iterations benchmark ====
Timings for HTTP JSON deserialization:
This package: 2.3955996250006137
Official SDK using pydantic model: 11.389213957998436
Official SDK using http model: 10.174893917006557

Timings for HTTP JSON serialization:
This package: 3.497491959002218
Official SDK using pydantic model: 31.92037604199868
Official SDK using http model: 6.780242209002608

Supported specification features

Core Specification	v1.0
CloudEvents Core	✅

---

Event Formats	v1.0
AVRO Event Format	❌
JSON Event Format	✅

---

Protocol Bindings	v1.0
HTTP Protocol Binding	✅
Kafka Protocol Binding	❌

---

Content Modes	v1.0
HTTP Binary	✅
HTTP Structured	✅
HTTP Batch	✅
Kafka Binary	❌
Kafka Structured	❌
Kafka Batch	❌

Commands for development

All the common commands used during development can be run using make targets:

• make dev-dependencies: Install dev requirements
• make fix: Run code style and lint automatic fixes (where possible)
• make test: Run test suite against system python version
• make check: Run tests against all available python versions, code style and lint checks
• make type, make format, make lint, make bandit: Run the relevant check
• make docs: Render the mkdocs website locally