Reference library

Data pipelines & processing

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

56 matches
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 easy

Pipeline stage compose functions left to right in Python

Compose multiple functions into a left-to-right pipeline so each stage receives the output of the previous one.

composition pipeline functional
Python
def compose(*funcs):
    """Compose functions left to right: compose(f, g, h)(x) == h(g(f(x)))"""
    def composed(arg):
        result = arg
        for func in funcs:
            result = func(result)
        return result
    return composed

if __name__ == "__main__":
    def add_one(x):
        return x + 1

    …
16 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 medium

Python Exponential Backoff Retry Example

Retry a flaky function with exponential backoff and jitter-free delays, printing each attempt and finally returning the successful result.

retry backoff exception-handling
Python
import random
import time


def flaky_function():
    if random.random() < 0.6:
        raise ConnectionError("Temporary network error")
    return "success"


def retry_with_exponential_backoff(func, max_retries=5, base_delay=1.0):
    for attempt in range(max_retries + 1):
        try:
            return func()
    …
16 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
Data pipelines & processing easy

Validate dict schema at pipeline boundary in Python

This code validates a dictionary against a TypedDict schema at a pipeline boundary, enforcing required fields and types with custom error messages.

validation dict typeddict
Python
from typing import Any, TypedDict


class Person(TypedDict):
    name: str
    age: int
    email: str


def validate_person(data: dict[str, Any]) -> Person:
    errors: list[str] = []

    if not isinstance(data.get("name"), str) or not data["name"].strip():
        errors.append("name must be a non-empty string")
  …
13 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.