runtimeThe nightly job, end to end
Assemble the capstone. The taxi package next to your program provides parse_trips(text) (→ (trips, rejected_lines)) and ZONES. Write nightly(csv_text, sink, state, batch_size): parse the export, split the trips into batches of batch_size in order, and for each batch from state["done"] on, build rows {"trip_id", "borough", "total"} (borough of the pickup zone, Unknown if missing) and call sink.write(rows), retrying a TimeoutError up to 3 attempts in total (no sleeping needed); after each successful write set state["done"]. Any other error propagates. Return {"loaded": trips parsed, "rejected": rejected line numbers, "batches_written": by this call, "retries": retries used}. Provided files sit next to your program (they aren't editor tabs): taxi/parse.py, taxi/zones.py and the export trips.csv; print(Path("taxi/parse.py").read_text(encoding="utf-8")) shows one's source.
nightly(csv_text, sink, state, batch_size)from taxi.parse import parse_trips and from taxi.zones import ZONEStrip_id, pickup_zone_id, dropoff_zone_id, trip_distance_miles, fare_amount, tip_amount and total_amount; ZONES maps a zone id to its borough{"trip_id": int, "borough": str, "total": float} per trip, in the trips' order: borough from the trip's pickup_zone_id, total is its total_amountstate["done"] are skipped; state["done"] moves only after a write succeeds{"loaded", "rejected", "batches_written", "retries"}