Reference library

Data pipelines & processing

ETL-style flows, batch transforms, validation, and moving data between formats.

3 matches

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.

cdc changelog mock-data

Python

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):
       …

15 0 Open

Data pipelines & processing easy

How to create a dated snapshot path for a dataset in Python

Generate a versioned directory path combining a base directory, dataset name, and today's date, ready for creating snapshots in data pipelines.

date pathlib datasets

Python

import datetime
import os
from pathlib import Path


def snapshot_path(base_dir: str, dataset_name: str) -> Path:
    """Return a dated snapshot path for a dataset under a base directory."""
    today = datetime.date.today().isoformat()
    return Path(base_dir) / dataset_name / today


if __name__ == "__main__":
    …

15 0 Open

Data pipelines & processing easy

Rollback dataset to previous snapshot pointer in Python

A SnapshotManager class stores timestamped data snapshots and rolls back to the most recent snapshot at or before a target time.

snapshots rollback datetime

Python

from datetime import datetime, timedelta


class SnapshotManager:
    def __init__(self):
        self.snapshots = {}  # timestamp -> data
        self.current_pointer = None

    def create_snapshot(self, data):
        timestamp = datetime.now()
        self.snapshots[timestamp] = data
        self.current_pointer =…

13 0 Open

Browse by section

Each section groups closely related Python snippets.