Reference library

Data pipelines & processing

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

53 matches

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

Trigger a Pipeline When a New File Appears in a Directory

Poll a directory every 0.5 seconds and return the name of the first new file that appears, or None after a timeout.

polling filesystem file-watcher

Python

import time
from pathlib import Path


def watch_for_file(directory: str, interval: float = 0.5, timeout: float = 10.0) -> str | None:
    """Poll a directory and trigger when a new file appears."""
    watch_dir = Path(directory)
    watch_dir.mkdir(exist_ok=True)
    
    known_files = set(watch_dir.iterdir())
    s…

13 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.