lesson 246 Production Readiness Capstone · advanced

Review the design of a small application

The capstone application is a nightly fare report over NYC taxi trips. Its first version works, as one long function. A design review asks of each part: what is its one job, what does it depend on, and can it be tested without the rest? The answer is a pure core of calculations with reading and writing at the edges.

You'll be able to

Spot the responsibilities tangled in one function: reading, validating, computing, formatting, writing
Separate them so the core takes plain data and returns results, with no files or printing inside
Wire the parts together in a thin run function, across modules of a package

The idea

This module takes one realistic program through a production review, using everything from the course. The program: each night, read the day's taxi trips (an export from the D06 dataset), check them, total the fares per borough, and write a report the dashboard reads.

A design review looks at structure before details. The Software Architecture module introduced the principles; here they are applied as questions a reviewer asks about real code. *What is each function's one job?* A function that reads a file, validates rows, adds up money and prints has five reasons to change and no way to test one without the others. *What does the core depend on?* Calculations should take plain data and return plain data: a pure core (see Functional & Dynamic Python) can be tested with three hand-written trips, reused by the API and the batch job, and profiled on its own. *Where are the edges?* Reading files, the clock, printing and writing reports belong in a thin outer layer that wires the steps together.

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · Version one: everything in one function

import csv

with open("trips.csv", "w", encoding="utf-8") as f:
    f.write("trip_id,pickup_zone_id,dropoff_zone_id,trip_distance_miles,fare_amount,tip_amount,total_amount\n")
    f.write("16126,186,179,9.2,40.8,10.7,64.19\n16127,264,48,0.9,7.2,2.55,15.5\n16128,39,89,3.17,17.2,0,18.7\n")
    f.write("16130,263,236,0.5,5.1,1,11.1\n16131,164,142,2.81,16.3,4.41,26.46\n16132,70,148,10.43,42.9,11.12,66.71\n")
    f.write("16133,114,238,5.2,22.6,5.65,34\n16134,48,230,1.1,7.9,3.4,17.05\n")

ZONES = {39: "Brooklyn", 48: "Manhattan", 70: "Queens", 114: "Manhattan", 164: "Manhattan", 186: "Manhattan", 263: "Manhattan", 264: "Unknown"}

def nightly_report():
    revenue = {}
    with open("trips.csv", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            total = float(row["total_amount"])
            if total < 0:
                continue
            borough = ZONES.get(int(row["pickup_zone_id"]), "Unknown")
            revenue[borough] = revenue.get(borough, 0) + total
    for borough in sorted(revenue):
        print(f"{borough:10} {revenue[borough]:8.2f}")

nightly_report()

Now try to test the arithmetic: does a refund (a negative total) get skipped? With version one, every test must write a file called trips.csv into the current folder and read the printed text back. Run it anywhere else and it can't even start:

example-2.py · The calculation can't run without the file

import csv

ZONES = {186: "Manhattan"}

def nightly_report():
    revenue = {}
    with open("trips.csv", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            borough = ZONES.get(int(row["pickup_zone_id"]), "Unknown")
            revenue[borough] = revenue.get(borough, 0) + float(row["total_amount"])
    return revenue

# A test of the money rule, in a folder with no trips.csv:
print(nightly_report())

example-3.py · Version two: parse, compute and report as separate steps

import csv
import io

ZONES = {39: "Brooklyn", 48: "Manhattan", 70: "Queens", 114: "Manhattan", 164: "Manhattan", 186: "Manhattan", 263: "Manhattan", 264: "Unknown"}

def parse_trips(text):
    """Text of the CSV export -> list of trips with typed fields."""
    return [
        {"pickup_zone_id": int(row["pickup_zone_id"]), "total_amount": float(row["total_amount"])}
        for row in csv.DictReader(io.StringIO(text))
    ]

def borough_revenue(trips, zones):
    """Pure core: trips and a zone table in, revenue per borough out."""
    revenue = {}
    for trip in trips:
        if trip["total_amount"] < 0:
            continue
        borough = zones.get(trip["pickup_zone_id"], "Unknown")
        revenue[borough] = round(revenue.get(borough, 0) + trip["total_amount"], 2)
    return revenue

def format_report(revenue):
    return [f"{borough:10} {revenue[borough]:8.2f}" for borough in sorted(revenue)]

def run(path):
    """Edge: read the file, call the core, print the report."""
    with open(path, encoding="utf-8") as f:
        trips = parse_trips(f.read())
    for line in format_report(borough_revenue(trips, ZONES)):
        print(line)

# The core is testable with hand-made data: no file, no printing.
print(borough_revenue([{"pickup_zone_id": 186, "total_amount": 64.19}, {"pickup_zone_id": 186, "total_amount": -12.5}], ZONES))

with open("trips.csv", "w", encoding="utf-8") as f:
    f.write("trip_id,pickup_zone_id,total_amount\n16126,186,64.19\n16128,39,18.7\n16132,70,66.71\n16133,114,34\n")
run("trips.csv")

In a real repository the steps live in modules of a package: taxi/parse.py, taxi/zones.py, taxi/summary.py, and a main.py (or a CLI, the CLI & Automation module) that wires them. The second challenge works that way: the package is provided as files next to your program.

Review checklist: one job per function; the core takes data, not file names; no print or open inside calculations; settings and tables passed in, not read from globals; imports point inward (the core imports nothing from the edges). Mistakes to avoid: splitting into functions that still share hidden state; a "utils" module of unrelated helpers; and moving code around without tests to show the behaviour didn't change (next lesson).

Review

Ask of every function: what is its one job, and can it run without files, printing or globals?
A pure core takes plain data (and tables or settings as parameters) and returns results; the edges read, write and wire.
In a package, the edge (run/main) imports the core modules, never the other way round.
Next: with the parts separated, test the whole pipeline: its outputs and its properties.

Practise this