Event Envelope with Schema Version Field in Python

Build a typed event envelope dataclass with an explicit schema version field for mock streaming scenarios.

Easy Python 3.7+ Aug 9, 2026 Streaming & messaging 15 views 0 copies

event dataclass messaging schema mock streaming

Python code

21 lines

Python 3.7+

from dataclasses import dataclass, field
from datetime import datetime
import uuid


@dataclass
class Event:
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    event_type: str = "user.created"
    version: str = "1.0.0"
    created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
    payload: dict = field(default_factory=dict)


if __name__ == "__main__":
    event = Event(payload={"user": "alice", "email": "alice@example.com"})
    print(f"event_id: {event.event_id}")
    print(f"event_type: {event.event_type}")
    print(f"version: {event.version}")
    print(f"created_at: {event.created_at}")
    print(f"payload: {event.payload}")

Output

stdout

event_id: 123e4567-e89b-12d3-a456-426614174000
event_type: user.created
version: 1.0.0
created_at: 2024-01-15T10:30:00.000000Z
payload: {'user': 'alice', 'email': 'alice@example.com'}

How it works

The dataclass automatically generates default values for event_id and created_at via factory functions, so every Event instance has a unique ID and timestamp. The version field is set manually to '1.0.0', representing the schema version of the envelope. Using a dataclass makes the envelope self-documenting and easy to extend with validation or serialization methods. This pattern is ideal for mock events in testing or prototyping message-driven systems before real producers are wired up.

Common mistakes

Forgetting that datetime.utcnow() is deprecated in Python 3.12 — use datetime.now(timezone.utc) instead
Not adding a type annotation to the payload field, which breaks clarity
Missing the __if __name__ guard, which executes mock code on import

Variations

Add a to_dict() method for easy serialization to JSON before publishing to Kafka
Use a Literal type for version to lock it to allowed schema versions

Real-world use cases

Unit-testing downstream consumers with deterministic mock events before live data flows exist.
Mocking schema-versioned events in a local Kafka or RabbitMQ dev environment.
Prototyping new event types in design documents without standing up full infrastructure.

Event Envelope with Schema Version Field in 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