Data pipelines &amp; processing

import json
from pathlib import Path
from typing import Any, Dict, List


def load_json_file(filepath: str) -> Dict[str, Any]:
    """Load JSON data from a file."""
    with Path(filepath).open("r", encoding="utf-8") as file:
        return json.load(file)


def filter_by_key(
    data: List[Dict[str, Any]], key: str,…

data-enrichment dictionaries pipelines

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.

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…

How to Hash Email Addresses in a PII Masking Pipeline in Python

Replaces every email address in a text string with its SHA-256 hash to protect personally identifiable information (PII).

pii hashing sha256

import hashlib
import re

def hash_email(email: str) -> str:
    """Mask an email address by hashing it with SHA-256."""
    normalized = email.strip().lower()
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()

def mask_pii_emails(text: str) -> str:
    """Replace all email addresses in text with their…

type-conversion robust-parsing data-cleaning

How to Safely Coerce Strings to Numbers in Python

A safe conversion function that turns strings into integers or floats, returning a fallback value when conversion fails.

import math

def to_number(value, fallback=None):
    """Safely coerce a string to int or float, returning fallback on failure."""
    if isinstance(value, (int, float)):
        return value
    try:
        # Try int first for clean whole numbers
        return int(value)
    except (ValueError, TypeError):
        …

12 0 Open

threads threadpoolexecutor concurrency

Parallel Extract Multiple Sources with Threads in Python

Extract data from multiple sources in parallel using ThreadPoolExecutor and verify results match sequential processing.

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…