Reference library

Data pipelines & processing

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

5 matches

Fan Out Records to Multiple Sinks in Python

Distribute the same records across multiple target sinks (database, API, queue, etc.) using a defaultdict-based fan-out pattern.

fan-out defaultdict records

Python

import json
from collections import defaultdict

SINKS = ["database", "api", "message_queue", "data_lake", "monitoring"]

def fan_out(records, *sinks):
    dist = defaultdict(list)
    for record in records:
        for sink in sinks:
            dist[sink].append(record)
    return dict(dist)

if __name__ == "__main_…

13 0 Open

Data pipelines & processing easy

How to Implement a Sliding Window Average in Python

Compute the average of the most recent N values in a stream using a bounded deque, efficiently updating the total as new values arrive.

deque sliding-window streaming

Python

from collections import deque


class SlidingWindowAverage:
    def __init__(self, window_size):
        self.window_size = window_size
        self.window = deque(maxlen=window_size)
        self.total = 0

    def add(self, value):
        if len(self.window) == self.window_size:
            self.total -= self.windo…

15 0 Open

Data pipelines & processing easy

How to Validate Data in a Python Pipeline

A helper module to validate common record types — email, positive integer, and non-empty string list — before processing data in a pipeline.

data-validation pipelines type-checking

Python

from typing import Any, Iterable


def is_valid_email(email: str) -> bool:
    """Basic email check: one '@', no spaces, dot after '@'."""
    if "@" not in email or " " in email:
        return False
    local, _, domain = email.partition("@")
    return bool(local) and "." in domain


def is_positive_int(value: Any)…

12 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 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

Browse by section

Each section groups closely related Python snippets.