Batch pipelines with Python

order_id,customer_name,item_id,qty,created_at
1001,Asha,cappuccino,2,2026-10-01 08:15:00
1001,Asha,croissant,1,2026-10-01 08:15:00
1002,Ravi,filter-coffee,1,2026-10-01 09:02:00
1003,Meena,cold-brew,1,2026-10-01 11:40:00
1004,Kiran,pizza,1,2026-10-01 12:10:00
1005,Divya,masala-chai,abc,2026-10-01 13:00:00
1006,Arjun,espresso,3,2026-10-01 18:30:00
1006,Arjun,espresso,3,2026-10-01 18:30:00

"""Batch pipeline: raw order CSV -> validated rows -> SQLite warehouse."""
import argparse
import csv
import logging
import sqlite3
from datetime import datetime
from pathlib import Path

MENU = {
    "espresso": 120, "cappuccino": 180, "cold-brew": 220, "filter-coffee": 80,
    "masala-chai": 90, "croissant": 150, "banana-bread": 130, "veg-sandwich": 160,
}

log = logging.getLogger("pipeline")


def extract(path):
    """Read the raw CSV. Everything is still text at this point."""
    with open(path, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def validate(row):
    """Return (clean_row, None) or (None, reason)."""
    try:
        order_id = int(row["order_id"])
        qty = int(row["qty"])
        created_at = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S")
    except (ValueError, KeyError) as exc:
        return None, f"bad value: {exc}"
    item_id = row["item_id"].strip()
    if item_id not in MENU:
        return None, f"unknown item_id: {item_id}"
    if qty <= 0:
        return None, "qty must be positive"
    clean = {
        "order_id": order_id,
        "item_id": item_id,
        "qty": qty,
        "line_total": qty * MENU[item_id],
        "order_date": created_at.date().isoformat(),
    }
    return clean, None


def transform(rows):
    """Split rows into good and rejected, and drop exact duplicates."""
    good, rejected, seen = [], [], set()
    for row in rows:
        clean, reason = validate(row)
        if clean is None:
            rejected.append({**row, "reason": reason})
            continue
        key = (clean["order_id"], clean["item_id"])
        if key in seen:
            rejected.append({**row, "reason": "duplicate line"})
            continue
        seen.add(key)
        good.append(clean)
    return good, rejected


def load(conn, rows, run_date):
    """Idempotent load: replace the whole day inside one transaction."""
    with conn:  # commits on success, rolls back on error
        conn.execute(
            """CREATE TABLE IF NOT EXISTS order_lines (
                   order_id INTEGER, item_id TEXT, qty INTEGER,
                   line_total INTEGER, order_date TEXT,
                   PRIMARY KEY (order_id, item_id))"""
        )
        conn.execute("DELETE FROM order_lines WHERE order_date = ?", (run_date,))
        conn.executemany(
            """INSERT INTO order_lines
               VALUES (:order_id, :item_id, :qty, :line_total, :order_date)""",
            rows,
        )


def write_rejects(rejected, path):
    if not rejected:
        return
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=list(rejected[0].keys()))
        writer.writeheader()
        writer.writerows(rejected)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--date", required=True, help="YYYY-MM-DD")
    parser.add_argument("--db", default="warehouse.db")
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

    path = Path("data") / f"orders_{args.date}.csv"
    raw = extract(path)
    good, rejected = transform(raw)
    log.info("read=%d good=%d rejected=%d", len(raw), len(good), len(rejected))

    if raw and len(rejected) / len(raw) > 0.5:
        raise SystemExit("More than half the rows were rejected; not loading.")

    write_rejects(rejected, Path("data") / f"rejects_{args.date}.csv")
    conn = sqlite3.connect(args.db)
    try:
        load(conn, good, args.date)
    finally:
        conn.close()
    log.info("loaded %d rows for %s", len(good), args.date)


if __name__ == "__main__":
    main()

python pipeline.py --date 2026-10-01

import sqlite3

conn = sqlite3.connect("warehouse.db")
print("rows:", conn.execute("SELECT COUNT(*) FROM order_lines").fetchone()[0])
for row in conn.execute(
    "SELECT order_date, SUM(line_total) FROM order_lines GROUP BY order_date"
):
    print(row)
conn.close()

python check.py
python pipeline.py --date 2026-10-01
python check.py

from pipeline import transform, validate


def test_validate_accepts_good_row():
    row = {"order_id": "1", "customer_name": "Asha", "item_id": "espresso",
           "qty": "2", "created_at": "2026-10-01 08:15:00"}
    clean, reason = validate(row)
    assert reason is None
    assert clean["line_total"] == 240
    assert clean["order_date"] == "2026-10-01"


def test_validate_rejects_unknown_item():
    row = {"order_id": "1", "customer_name": "Asha", "item_id": "pizza",
           "qty": "1", "created_at": "2026-10-01 08:15:00"}
    clean, reason = validate(row)
    assert clean is None
    assert "unknown item_id" in reason


def test_transform_drops_duplicates():
    row = {"order_id": "1", "customer_name": "Asha", "item_id": "espresso",
           "qty": "1", "created_at": "2026-10-01 08:15:00"}
    good, rejected = transform([row, dict(row)])
    assert len(good) == 1
    assert rejected[0]["reason"] == "duplicate line"

You are a senior data engineer. Review this Python batch pipeline for production readiness. Look for: non-idempotent loads, missing error handling, silent data loss, memory problems on large files, hard-coded values and secrets, and missing tests. Give a prioritised list of fixes with short code examples.

[paste your pipeline code here]