P17 Validate and quarantine · P18 Diff two snapshots

FreeLesson 66 of 78

P17Validate and quarantineDEMEDIUM

Sounds like: "reject bad rows but load the rest", "enforce a data contract", "where do bad records go?"

Worked question: Check every row of orders_dirty.csv against the rules, load the good ones, and park the bad ones with reasons.

Approach: 1. one function returns the cleaned row and a list of errors 2. collect every error, not just the first 3. two outputs: good and quarantine

SOLUTIONPYTHON 3.12

import csv
from datetime import date

ALLOWED = {"delivered", "cancelled", "returned"}

def validate(row):
    errs = []
    if not row["customer_id"].strip():
        errs.append("customer_id missing")
    try:
        date.fromisoformat(row["order_date"])
    except ValueError:
        errs.append(f"bad date {row['order_date']}")
    status = row["status"].strip().lower()
    if status not in ALLOWED:
        errs.append(f"unknown status {status!r}")
    try:
        if float(row["amount"]) <= 0:
            errs.append(f"amount {row['amount']} <= 0")
    except ValueError:
        errs.append(f"amount not numeric {row['amount']!r}")
    return {**row, "status": status}, errs

good, quarantine = [], []
for row in csv.DictReader(open("orders_dirty.csv")):
    clean, errs = validate(row)
    (quarantine if errs else good).append((clean, errs))

print("good:", [(r["order_id"], r["status"]) for r, _ in good])
for r, errs in quarantine:
    print("bad ", r["order_id"], errs)

Output

good: [('121', 'delivered'), ('122', 'delivered')]
bad  116 ['amount -50 <= 0']
bad  117 ['customer_id missing']
bad  118 ['bad date 2025-02-30']
bad  119 ["unknown status 'shipped'"]
bad  120 ["amount not numeric 'abc'"]

P18Diff two snapshotsDEMEDIUM

Sounds like: "what changed since yesterday", "build a CDC feed from daily dumps", "feed an SCD Type 2 load"

Worked question: Compare two daily customer snapshots and emit inserts, deletes and field-level updates.

Approach: 1. key both snapshots by id 2. set algebra on the keys 3. compare fields for ids in both

SOLUTIONPYTHON 3.12

import csv

def load(path):
    return {r["customer_id"]: r for r in csv.DictReader(open(path))}

old, new = load("snapshot_day1.csv"), load("snapshot_day2.csv")

inserts = sorted(new.keys() - old.keys())
deletes = sorted(old.keys() - new.keys())
updates = []
for k in sorted(old.keys() & new.keys(), key=int):
    changed = {c: (old[k][c], new[k][c])
               for c in new[k] if old[k][c] != new[k][c]}
    if changed:
        updates.append((k, changed))

print("inserts:", inserts, " deletes:", deletes)
for k, ch in updates:
    print("update ", k, ch)

Output

inserts: ['9']  deletes: ['8']
update  4 {'city': ('Bengaluru', 'Pune')}
update  6 {'city': ('', 'Chennai')}

Practice before you move on

Solve the two practice questions for P17 and P18 in Coding Practice, under Python.
Add a rule to P17 that rejects an order_date later than today.

In Coding Practice

Solve this pattern on fresh data, under Python. Hidden tests grade it.