How to Build a Mock Change Data Capture Event Stream in Python

from itertools import count
from random import choice, randint, seed
from datetime import datetime, timedelta

seed(42)  # Make output deterministic
event_types = ["INSERT", "UPDATE", "DELETE"]
table_names = ["users", "orders", "products", "payments"]
counter = count(1)

def mock_cdc_event(stream_index: int) -> dict:
    """Generate a single mock CDC event."""
    timestamp = datetime.now() + timedelta(seconds=randint(0, 86400))
    return {
        "event_id": next(counter),
        "stream_position": stream_index,
        "event_type": choice(event_types),
        "table_name": choice(table_names),
        "record_id": randint(1, 10000),
        "occurred_at": timestamp.isoformat(timespec="seconds"),
        "payload": {
            "old_value": None if choice([True, False]) else randint(0, 100),
            "new_value": randint(0, 100)
        }
    }

def event_stream_mock_list(events_count: int = 5) -> list[dict]:
    """Return a list of mock change data capture events."""
    return [mock_cdc_event(i) for i in range(1, events_count + 1)]

if __name__ == "__main__":
    for event in event_stream_mock_list():
        print(event)