Reference library

Data pipelines & processing

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

5 matches

Enrich Events with Geo IP Data in Python

Returns a copy of each event dictionary, enriched with a geo-location dict from a mock IP-to-geo lookup table, with a fallback for unknown IPs.

data-enrichment dictionaries pipelines

Python

import ipaddress


GEO_IP_DB = {
    "192.168.1.10": {"country": "US", "city": "New York", "lat": 40.7128, "lon": -74.0060},
    "10.0.0.5": {"country": "DE", "city": "Berlin", "lat": 52.5200, "lon": 13.4050},
    "172.16.0.8": {"country": "JP", "city": "Tokyo", "lat": 35.6762, "lon": 139.6503},
}

EVENTS = [
    {"id…

14 0 Open

Data pipelines & processing easy

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 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 Run a Mock Cron Pipeline Scheduler in Python

This code schedules a mock pipeline job to run every 2 seconds and hourly at :30 using the schedule library, then runs pending tasks for 10 seconds.

schedule cron pipeline

Python

import time
import schedule
from datetime import datetime


def run_pipeline():
    print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - Pipeline executed")


schedule.every(2).seconds.do(run_pipeline)
schedule.every().hour.at(":30").do(run_pipeline)

print("Scheduler started. Press Ctrl+C to stop.")
end_time = ti…

13 0 Open

Data pipelines & processing easy

Implement Exactly-Once Transaction Log in Python

A mock transaction log that deduplicates transaction IDs so each is recorded only once, with a dataclass for records and simple in-memory storage.

transactions deduplication dataclass

Python

from dataclasses import dataclass
from typing import Dict, Optional


@dataclass
class TxnRecord:
    txn_id: str
    status: str


class ExactlyOnceTxnLog:
    def __init__(self) -> None:
        self._log: Dict[str, TxnRecord] = {}
        self._processed_ids: set = set()

    def record(self, txn_id: str, status: s…

14 0 Open

Browse by section

Each section groups closely related Python snippets.