Reference library

Data pipelines & processing

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

6 matches

Extract Schema.org Structured Data from Any Website in Python

A Python tool that fetches a webpage and extracts all JSON-LD structured data (Schema.org) embedded in <script> tags with type="application/ld+json".

web-scraping structured-data schema-org

Python

import requests
from bs4 import BeautifulSoup
import json

def extract_schema_org(url):
    """Extract structured data (Schema.org) from a website."""
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        return {"err…

51 0 Open

Data pipelines & processing easy

Filter Records by Required Fields in Python

Filter a list of dictionaries, keeping only records where every required field is present and not None.

filter data-cleaning pipelines

Python

def filter_records(records, required_fields):
    """Return only records that have all required fields non-null."""
    return [
        record for record in records
        if all(record.get(field) is not None for field in required_fields)
    ]


if __name__ == "__main__":
    sample_records = [
        {"name": "Al…

14 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

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

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

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.