Reference library

Data pipelines & processing

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

5 matches

Create Data Helper Functions in Python for Beginners

Build reusable Python helper functions to load, filter, sort, summarize, and save JSON data — a beginner-friendly starting point for small data pipelines.

json pipeline helpers

Python

import json
from pathlib import Path
from typing import Any, Dict, List


def load_json_file(filepath: str) -> Dict[str, Any]:
    """Load JSON data from a file."""
    with Path(filepath).open("r", encoding="utf-8") as file:
        return json.load(file)


def filter_by_key(
    data: List[Dict[str, Any]], key: str,…

14 0 Open

Data pipelines & processing easy

How to Build Data Processing Functions in Python

Create reusable helper functions to load, filter, transform, and aggregate CSV data in Python.

csv pipeline etl

Python

import csv
from pathlib import Path


def load_data(filepath):
    """Load CSV data into a list of dicts."""
    with open(filepath, "r", newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def filter_rows(rows, column, value):
    """Keep rows where column equals value."""
    return [row for…

11 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

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

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

Browse by section

Each section groups closely related Python snippets.