Reference library

Data pipelines & processing

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

56 matches
Data pipelines & processing easy

Idempotent Pipeline Dedupe by Record ID Set in Python

Filters records against a persistent set of seen IDs, returning only new ones and the updated set for idempotent pipeline processing.

deduplication idempotency pipelines
Python
def dedupe_records(records, seen_ids=None):
    """Return records whose id has not been seen before."""
    if seen_ids is None:
        seen_ids = set()
    unique = []
    for record in records:
        record_id = record.get("id")
        if record_id not in seen_ids:
            seen_ids.add(record_id)
           …
12 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
Data pipelines & processing medium

Map Partition Over Chunks in Python with Multiprocessing and Mock

Process data in chunks across multiple CPU cores using multiprocessing Pool.map, and mock the chunk function to test partitioning behavior without heavy computation.

multiprocessing chunking parallel
Python
from multiprocessing import Pool
from unittest.mock import patch, Mock

def process_chunk(chunk):
    return [x * x for x in chunk]

def map_partition_over_chunks(data, chunk_size, process_func=process_chunk):
    chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
    with Pool() as pool:
     …
12 0 Open
Data pipelines & processing easy

Parallel Extract Multiple Sources with Threads in Python

Extract data from multiple sources in parallel using ThreadPoolExecutor and verify results match sequential processing.

threads threadpoolexecutor concurrency
Python
import threading
from concurrent.futures import ThreadPoolExecutor

def extract_from_source(source):
    """Simulate extracting data from a source."""
    return f"Data from {source}"

def main():
    sources = ["source_a", "source_b", "source_c", "source_d"]
    
    # Sequential extraction for comparison
    sequent…
14 0 Open
Data pipelines & processing medium

Pivot long to wide transformation dict

Transform a list of dictionaries from long format to wide format by pivoting on a key column and aggregating values, using pure Python.

pivot transformation data-cleaning
Python
def pivot_long_to_wide(rows, key_col, value_col, id_cols=None):
    """
    Convert long-format data (list of dicts) to wide format.
    
    Args:
        rows: List of dicts in long format
        key_col: Column name to pivot on (becomes new column headers)
        value_col: Column name whose values become the cel…
11 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
Data pipelines & processing easy

Test a Python Pipeline with Fixture Sample Rows

Test pipeline functions with sample rows provided by a pytest fixture, verifying required keys and value constraints.

pytest fixtures data-pipelines
Python
import pytest


def get_value(data: dict, key: str):
    return data.get(key)


def sample_rows():
    return [
        {"name": "Alice", "age": 30, "city": "London"},
        {"name": "Bob", "age": 25, "city": "Paris"},
        {"name": "Charlie", "age": 35, "city": "Berlin"},
    ]


@pytest.fixture
def sample_data(…
16 0 Open
Data pipelines & processing easy

Union Multiple DataFrames with Aligned Columns in Python

Concatenate DataFrames with different columns, aligning them and filling missing values with NaN using pandas concat.

pandas dataframes concat
Python
import pandas as pd
from io import StringIO

# Sample dataframes with different columns
df1 = pd.DataFrame({
    'id': [1, 2, 3],
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [25, 30, 35]
})

df2 = pd.DataFrame({
    'id': [4, 5],
    'name': ['Diana', 'Eve'],
    'city': ['NYC', 'LA']
})

df3 = pd.DataFrame({
…
14 0 Open

Browse by section

Each section groups closely related Python snippets.

Data pipelines & processing — Python code examples

What you will find here

This page collects data pipelines & processing snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.

Samples vs tutorials and challenges

Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.