Reference library

Data pipelines & processing

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

5 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…

50 0 Open

Data pipelines & processing medium

How to Stream a Large JSONL File Line by Line in Python

Process a large JSON-lines file incrementally using streaming techniques to avoid loading the entire file into memory.

streaming jsonl large-files

Python

import json

def process_large_file(filepath, chunk_size=8192):
    """
    Stream a large JSON-lines file line by line, processing each record
    without loading the entire file into memory.
    """
    total_count = 0
    total_sum = 0
    
    with open(filepath, 'r') as f:
        while True:
            chunk = …

13 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

Map Partition Over Chunks in Python with Multiprocessing and Mock

Process data in chunks across multiple CPU cores using multiprocessing Pool.map, and mock the chunk function to test partitioning behavior without heavy computation.

multiprocessing chunking parallel

Python

from multiprocessing import Pool
from unittest.mock import patch, Mock

def process_chunk(chunk):
    return [x * x for x in chunk]

def map_partition_over_chunks(data, chunk_size, process_func=process_chunk):
    chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
    with Pool() as pool:
     …

12 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

Browse by section

Each section groups closely related Python snippets.