Reference library

Data pipelines & processing

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

4 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 Sort a List of Dictionaries by Key in Python

A reusable helper function that sorts a list of dictionaries by a specified key, with optional descending order support.

sorting dictionaries data-pipelines

Python

from typing import List

def sort_records(records: List[dict], key: str, descending: bool = False) -> List[dict]:
    """Sort a list of dictionaries by a specified key."""
    return sorted(records, key=lambda record: record[key], reverse=descending)


def demonstrate_sorting() -> None:
    users = [
        {"name": …

12 0 Open

Data pipelines & processing medium

How to Topologically Sort a DAG in Python

Compute a valid execution order for tasks with dependencies using Kahn's algorithm in Python.

dag topological-sort graph

Python

from collections import defaultdict, deque


def topological_order(dependencies):
    graph = defaultdict(list)
    in_degree = defaultdict(int)
    tasks = set(dependencies.keys())

    for task, depends_on in dependencies.items():
        for d in depends_on:
            graph[d].append(task)
            in_degree[t…

11 0 Open

Data pipelines & processing medium

Implement an Out-of-Order Sort Buffer with a Heap in Python

Buffers out-of-order indices from a stream and emits them in sorted order using a min-heap with a sliding window.

heapq sorting streaming

Python

import heapq
from collections import deque


class OutOfOrderSorter:
    def __init__(self, buffer_size):
        self.buffer_size = buffer_size
        self.buffer = deque(maxlen=buffer_size)
        self.heap = []
        self.next_expected_index = 0
        self.result = []

    def push(self, item):
        heapq.…

12 0 Open

Browse by section

Each section groups closely related Python snippets.