Data pipelines &amp; processing

from pathlib import Path
import json

def attach_source_metadata(records, source_file):
    """Attach source filename metadata to each record."""
    return [
        {**record, "source": Path(source_file).name}
        for record in records
    ]

if __name__ == "__main__":
    source = "/data/raw/customers.csv"
    …

18 0 Open

ETL in Python: Extract CSV, Transform Dict, Load JSON

Build a simple ETL pipeline in Python that reads a CSV file, transforms each row (stripping whitespace and converting numeric fields), and writes the result to JSON.

etl csv json

import csv
import json
from pathlib import Path

def extract_csv(file_path):
    """Read CSV file and return list of row dictionaries."""
    with Path(file_path).open('r', newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        return list(reader)

def transform_dicts(rows):
    """Transform ro…

14 0 Open

ETL in Python: Extract CSV, Transform Dicts, Load JSON

Build a simple ETL pipeline that reads a CSV, normalizes keys and converts price to float, then writes structured JSON.

etl csv json

import csv
import json
from pathlib import Path

def etl_csv_to_json(csv_path: str, json_path: str) -> None:
    """Extract CSV, transform rows to dicts, load to JSON."""
    with open(csv_path, mode='r', newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        records = list(reader)

    # Trans…

How to Build Data Processing Functions in Python

Create reusable helper functions to load, filter, transform, and aggregate CSV data in Python.

csv pipeline etl

import csv
from pathlib import Path


def load_data(filepath):
    """Load CSV data into a list of dicts."""
    with open(filepath, "r", newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def filter_rows(rows, column, value):
    """Keep rows where column equals value."""
    return [row for…

incremental-load watermark sqlite

How to Implement Incremental Load with Watermark by updated_at in Python

Load only new or changed rows into SQLite by comparing an updated_at timestamp against a stored watermark, returning counts and the new watermark.

import sqlite3
from datetime import datetime, timedelta


def watermark_incremental_load(db_path, table_name, last_watermark, source_data):
    """Load only rows with updated_at greater than the last watermark."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # Create table if it doesn't exist
  …

data-validation pipelines type-checking

How to Validate Data in a Python Pipeline

A helper module to validate common record types — email, positive integer, and non-empty string list — before processing data in a pipeline.

from typing import Any, Iterable


def is_valid_email(email: str) -> bool:
    """Basic email check: one '@', no spaces, dot after '@'."""
    if "@" not in email or " " in email:
        return False
    local, _, domain = email.partition("@")
    return bool(local) and "." in domain


def is_positive_int(value: Any)…