How to Wrap Message Attributes in a CloudEvent with Python

Create a minimal CloudEvent dataclass that wraps arbitrary message attributes into a JSON envelope, matching CloudEvents 1.0 spec.

Easy Python 3.9+ Aug 9, 2026 Streaming & messaging 13 views 0 copies

cloudevents messaging dataclasses json mock

Python code

34 lines

Python 3.9+

import json
from dataclasses import dataclass, field, asdict
from typing import Any, Dict
from datetime import datetime, timezone


@dataclass
class CloudEvent:
    message_attributes: Dict[str, Any] = field(default_factory=dict)

    def wrap(self, event_id: str, source: str, event_type: str, data: Any):
        self.message_attributes = {
            "specversion": "1.0",
            "id": event_id,
            "source": source,
            "type": event_type,
            "datacontenttype": "application/json",
            "time": datetime.now(timezone.utc).isoformat(),
            "data": data,
        }
        return self

    def to_json(self) -> str:
        return json.dumps(asdict(self)["message_attributes"], indent=2)


if __name__ == "__main__":
    event = CloudEvent().wrap(
        event_id="1234-5678",
        source="/mock/orders",
        event_type="order.created",
        data={"order_id": "A-1001", "status": "pending"},
    )
    print(event.to_json())

Output

stdout

{
  "specversion": "1.0",
  "id": "1234-5678",
  "source": "/mock/orders",
  "type": "order.created",
  "datacontenttype": "application/json",
  "time": "2025-01-06T12:34:56.789012+00:00",
  "data": {
    "order_id": "A-1001",
    "status": "pending"
  }
}

How it works

The CloudEvent dataclass provides a lightweight mock wrapper for CloudEvents message attributes. The wrap method populates the message_attributes dict with standard CloudEvents fields (specversion, id, source, type, datacontenttype, time) plus the payload under data. datetime.now(timezone.utc).isoformat() generates a timestamp in ISO-8601 format with a UTC offset, which matches CloudEvents time format. asdict converts the dataclass to a plain dict, and json.dumps(..., indent=2) produces a readable JSON representation. Using a default_factory=dict ensures each instance starts with an independent empty dict rather than a shared mutable default.

Common mistakes

Forgetting `timezone.utc` in `datetime.now()` results in a naive timestamp without timezone info.
Using a mutable default like `{}` in the dataclass field causes shared state across instances.
Confusing `asdict(self)` with directly dumping the dataclass — `asdict` is needed to convert nested fields.

Variations

Use `pydantic` models to validate CloudEvent fields automatically.
Store attributes on the dataclass directly instead of nested `message_attributes`.

Real-world use cases

Mocking cloud event objects in unit tests to simulate messages published to Kafka or Pub/Sub.
Building a lightweight event wrapper in serverless functions before forwarding to a stream.
Creating consistent CloudEvent envelopes in a demo or PoC without pulling in heavyweight SDKs.

How to Wrap Message Attributes in a CloudEvent with Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Keep learning

Tutorials

Quizzes