Reference library

Data pipelines & processing

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

3 matches

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 medium

Enrich a stream with reference data by key lookup in Python

Uses streamz to join each incoming record to a reference dictionary by name, adding department and level fields or defaults.

streamz streaming join

Python

from streamz import Stream

reference = {"alice": {"dept": "eng", "level": 3}, "bob": {"dept": "sales", "level": 5}}

def enrich(record):
    name = record.get("name")
    ref = reference.get(name)
    joined = dict(record)
    if ref:
        joined.update(ref)
    else:
        joined["dept"] = "unknown"
        joi…

14 0 Open

Data pipelines & processing medium

How to perform a star schema join in Python

Denormalize mock fact and dimension tables by building lookup dicts and enriching each sales fact with customer, product, and date attributes.

star-schema data-joins dimensional-modeling

Python

from datetime import date

# Mock dimension tables
customers = [
    {"customer_id": 1, "name": "Alice", "city": "New York"},
    {"customer_id": 2, "name": "Bob", "city": "Los Angeles"},
    {"customer_id": 3, "name": "Carol", "city": "Chicago"},
]

products = [
    {"product_id": 101, "name": "Laptop", "category": "…

12 0 Open

Browse by section

Each section groups closely related Python snippets.