Generate a Mock CDC Changelog in Python

Simulate a CDC changelog with INSERT, UPDATE, and DELETE operations, timestamps, and record snapshots for testing data pipelines.

Easy Python 3.9+ Aug 9, 2026 Data pipelines & processing 15 views 0 copies

cdc changelog mock-data debezium testing

Python code

30 lines

Python 3.9+

import json
from datetime import datetime, timedelta


def generate_mock_changelog(records, operations=("INSERT", "UPDATE", "DELETE")):
    """Simulate a CDC changelog from a list of record snapshots."""
    base_time = datetime(2025, 1, 1, 8, 0, 0)
    changelog = []
    for idx, record in enumerate(records):
        change_time = base_time + timedelta(minutes=idx * 2)
        op = operations[idx % len(operations)]
        changelog.append(
            {
                "op": op,
                "ts_ms": int(change_time.timestamp() * 1000),
                "after": record if op != "DELETE" else None,
                "before": record if op == "DELETE" else {},
            }
        )
    return changelog


if __name__ == "__main__":
    sample_records = [
        {"id": 1, "name": "Alice", "email": "alice@example.com"},
        {"id": 2, "name": "Bob", "email": "bob@example.com"},
        {"id": 3, "name": "Carol", "email": "carol@example.com"},
    ]
    changelog = generate_mock_changelog(sample_records)
    print(json.dumps(changelog, indent=2))

Output

stdout

[
  {
    "op": "INSERT",
    "ts_ms": 1735725600000,
    "after": {
      "id": 1,
      "name": "Alice",
      "email": "alice@example.com"
    },
    "before": {}
  },
  {
    "op": "UPDATE",
    "ts_ms": 1735725720000,
    "after": {
      "id": 2,
      "name": "Bob",
      "email": "bob@example.com"
    },
    "before": {}
  },
  {
    "op": "DELETE",
    "ts_ms": 1735725840000,
    "after": null,
    "before": {
      "id": 3,
      "name": "Carol",
      "email": "carol@example.com"
    }
  }
]

How it works

This function cycles through a list of operation types using modulo indexing, so if you pass 4 records you'll get INSERT, UPDATE, DELETE, INSERT. Each entry gets a timestamp two minutes after the previous one, starting from a fixed base time. The after field holds the new record state for INSERT/UPDATE and is None for DELETE, while before holds the old state only for DELETE. This mirrors the format used by tools like Debezium and typical CDC consumers. The JSON output is deterministic, which makes it easy to write assertions in tests.

Common mistakes

Forgetting that DELETE events use `after: None` and put the record in `before`
Assuming timestamps are ISO strings instead of epoch milliseconds
Not cycling operations, so every record gets the same 'op' value
Hardcoding timestamps instead of computing them relative to a base time

Variations

Use `time.sleep` or a random offset to simulate more realistic event timing
Add a `source` key with table, database, and version metadata like Debezium does

Real-world use cases

Testing a consumer that reads from Kafka with Debezium-style CDC events without a live database.
Populating a local dev environment with fake database change events for debugging sync logic.
Writing unit tests for data validation logic that needs predictable INSERT/UPDATE/DELETE sequences.

Generate a Mock CDC Changelog in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Data pipelines & processing

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Data pipelines & processing

Keep learning

Tutorials

Quizzes