Reference library

Data pipelines & processing

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

4 matches
Data pipelines & processing easy

Enrich Events with Geo IP Data in Python

Returns a copy of each event dictionary, enriched with a geo-location dict from a mock IP-to-geo lookup table, with a fallback for unknown IPs.

data-enrichment dictionaries pipelines
Python
import ipaddress


GEO_IP_DB = {
    "192.168.1.10": {"country": "US", "city": "New York", "lat": 40.7128, "lon": -74.0060},
    "10.0.0.5": {"country": "DE", "city": "Berlin", "lat": 52.5200, "lon": 13.4050},
    "172.16.0.8": {"country": "JP", "city": "Tokyo", "lat": 35.6762, "lon": 139.6503},
}

EVENTS = [
    {"id…
14 0 Open
Data pipelines & processing easy

Group Python Events into Sessions with a Gap Timeout

Groups timestamped events into sessions, starting a new session when the time gap exceeds a specified timeout.

sessions grouping datetime
Python
from itertools import groupby
from datetime import datetime, timedelta

def session_window_group(events, gap_seconds=300):
    """Group events into sessions where gap > gap_seconds starts a new session."""
    if not events:
        return []
    
    events = sorted(events, key=lambda x: x[0])
    sessions = []
    c…
13 0 Open
Data pipelines & processing easy

How to Deduplicate Events with At-Least-Once Delivery in Python

Implements an exactly-once processing pattern for at-least-once event delivery by tracking seen event IDs in a set, skipping duplicates.

deduplication idempotent event-processing
Python
seen_ids = set()

def process_event(event_id: str, payload: dict) -> dict:
    """Process an event exactly once, ignoring duplicates."""
    if event_id in seen_ids:
        return {"status": "duplicate", "event_id": event_id}
    seen_ids.add(event_id)
    return {"status": "processed", "event_id": event_id, **payloa…
13 0 Open
Data pipelines & processing easy

How to route late-arriving data to a side output in Python

Separate late-arriving events from a streaming data batch into a dead-letter side output list using a timestamp threshold.

data pipelines streaming dead-letter
Python
from collections import defaultdict

def late_arriving_side_output(events, late_threshold_ts):
    """
    Mock a streaming pipeline that separates late-arriving data events
    into a side output list (e.g., for dead-letter analysis).

    events: list of (timestamp, data) tuples, timestamps as ints.
    late_thresho…
12 0 Open

Browse by section

Each section groups closely related Python snippets.

Data pipelines & processing — Python code examples

What you will find here

This page collects data pipelines & processing snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.

Samples vs tutorials and challenges

Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.