Reference library

Data pipelines & processing

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

6 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

Group Python Events into Sessions with a Gap Timeout

Groups timestamped events into sessions, starting a new session when the time gap exceeds a specified timeout.

sessions grouping datetime

Python

from itertools import groupby
from datetime import datetime, timedelta

def session_window_group(events, gap_seconds=300):
    """Group events into sessions where gap > gap_seconds starts a new session."""
    if not events:
        return []
    
    events = sorted(events, key=lambda x: x[0])
    sessions = []
    c…

13 0 Open

Data pipelines & processing easy

How to Implement Incremental Load with Watermark by updated_at in Python

Load only new or changed rows into SQLite by comparing an updated_at timestamp against a stored watermark, returning counts and the new watermark.

incremental-load watermark sqlite

Python

import sqlite3
from datetime import datetime, timedelta


def watermark_incremental_load(db_path, table_name, last_watermark, source_data):
    """Load only rows with updated_at greater than the last watermark."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # Create table if it doesn't exist
  …

12 0 Open

Data pipelines & processing easy

How to List Failed Records in a Dead Letter Queue Mock in Python

A mock Dead Letter Queue stores failed processing records with error details and timestamps, lists them, and exports to JSON.

dead-letter-queue json logging

Python

import json
from datetime import datetime, timedelta
import random


class DeadLetterQueue:
    def __init__(self):
        self.failed_records = []

    def add_failed_record(self, record_id, payload, error_message):
        self.failed_records.append({
            "record_id": record_id,
            "payload": paylo…

13 0 Open

Data pipelines & processing easy

How to route late-arriving data to a side output in Python

Separate late-arriving events from a streaming data batch into a dead-letter side output list using a timestamp threshold.

data pipelines streaming dead-letter

Python

from collections import defaultdict

def late_arriving_side_output(events, late_threshold_ts):
    """
    Mock a streaming pipeline that separates late-arriving data events
    into a side output list (e.g., for dead-letter analysis).

    events: list of (timestamp, data) tuples, timestamps as ints.
    late_thresho…

12 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.