Capstone: an automated reporting tool · PYTH-240 Lesson 12

Every lesson in this course has added a piece: HTTP and the shop API, pagination and client classes, authentication, the three data formats, polite scraping, sockets, spreadsheets, documents, email and notifications, command-line options and reliable scheduling. This capstone puts them together into the tool that started the course: Theo's Monday report. Today he spends most of Monday morning collecting numbers from the shop, the carrier and the importer and typing them into an email. By the end of this lesson the job runs by itself at 06:30, and Theo starts the week with the report already in his inbox, including a short list of the things that actually need his attention.

The lesson follows the way a professional would approach it: agree the requirements with the people who will use the result, design the pieces and their boundaries, build and test each piece against fake sources, assemble the command-line tool, then schedule and monitor it. The real sources need the network and Ridgeline's credentials, so here the tool runs against fake versions with the same interfaces, exactly as lesson 1 recommended. Every listing on this page is self-contained: each begins with the same small block of fake data for one week, followed by the piece being discussed.

Requirements

Jordan sat down with Theo, Kai and Anita for twenty minutes and wrote down what the report must contain and how it must behave. Writing requirements as short, checkable statements makes it obvious later whether the tool does its job.

What the weekly report must do
Requirement	Source	Lesson
Paid orders, refunds and sales total for the week, as exact money	Shop API, paginated JSON	2, 4
Bags sold per coffee, top three in the email, all in a CSV attachment	Shop API order lines	4, 7, 9
Parcels delivered, in transit and with problems; paid orders not yet shipped	Carrier tracking XML	4
Green coffee lots whose price moved by more than 0.50 per kg, new lots and lots no longer offered	Importer CSV feed, this week and last	4
Runs every Monday at 06:30 and emails Theo; never emails when run by hand unless asked	Scheduler and CLI options	10, 11
Survives a short shop outage; fails loudly if a source is down for longer	Retries, exit codes, heartbeat	11

Two requirements shaped the design more than the others. Theo asked for "only the things I need to act on" at the top, so the report marks action items clearly instead of listing every parcel. And Anita asked that the totals be exactly right, to the cent, because she copies them into the books, which rules out floats anywhere near the money.

Design: small pieces with clear boundaries

The tool is split into functions that each do one job, take plain data in and give plain data out. Fetching talks to the network and nothing else. Summarizing, delivery analysis and price comparison are pure calculations on data already fetched, so they can be tested with fixed inputs. Rendering turns results into text. Sending is a separate step that can be replaced by a dry run. The command-line main function wires them together and turns failures into exit codes. Each source is passed in as a parameter, so the same code runs against the fake week here and against the real services on the shop server.

The weekly report tool: fetch, analyze, render, send, all wired together by a command-line main

In the real project these functions live in a module, ridgeline/reports.py, inside the ridgeline-tools package, with the tests in tests/test_reports.py, following the layout from PYTH-102 and PYTH-110. The [project.scripts] entry from lesson 10 makes weekly-report a command.

Fetching the orders

fetch_orders combines lessons 2 and 11: it walks the shop's pages until the last one, and retries a page that fails with a timeout or connection error, waiting longer each time. The sleep function is a parameter so tests do not actually wait. The fake shop can be told to fail a number of times first, which lets the listing show the retries working.

Fetching every page of orders, with retries after two simulated timeouts

from decimal import Decimal

# ---- fake sources for one week (2 to 8 November 2026) ----
ORDERS = [
    {"id": 5101, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10101", "lines": [["RC-101", 2, "18.50"], ["RC-104", 1, "20.25"]]},
    {"id": 5102, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10102", "lines": [["RC-102", 1, "22.00"]]},
    {"id": 5103, "placed": "2026-11-03", "status": "refunded", "tracking": None, "lines": [["RC-101", 1, "18.50"]]},
    {"id": 5104, "placed": "2026-11-04", "status": "paid", "tracking": "RDG10104", "lines": [["RC-106", 2, "21.50"], ["RC-101", 1, "18.50"]]},
    {"id": 5105, "placed": "2026-11-05", "status": "paid", "tracking": "RDG10105", "lines": [["RC-103", 3, "19.75"]]},
    {"id": 5106, "placed": "2026-11-06", "status": "paid", "tracking": "RDG10106", "lines": [["RC-101", 4, "18.50"]]},
    {"id": 5107, "placed": "2026-11-07", "status": "paid", "tracking": "RDG10107", "lines": [["RC-104", 2, "20.25"]]},
    {"id": 5108, "placed": "2026-11-08", "status": "paid", "tracking": None, "lines": [["RC-102", 1, "22.00"], ["RC-105", 1, "19.00"]]},
]

class FakeShop:
    """Pages of three orders, like the real /orders endpoint."""
    def __init__(self, fail_first=0):
        self.fail_first = fail_first
    def get_orders(self, page):
        if self.fail_first:
            self.fail_first -= 1
            raise TimeoutError("shop API timed out")
        pages = (len(ORDERS) + 2) // 3
        return {"page": page, "pages": pages, "orders": ORDERS[(page - 1) * 3: page * 3]}

TRACKING_XML = """<tracking><parcel number="RDG10101" code="DLV"/><parcel number="RDG10102" code="DLV"/>
<parcel number="RDG10104" code="DLV"/><parcel number="RDG10105" code="TRN"/>
<parcel number="RDG10106" code="EXC" note="address not found"/><parcel number="RDG10107" code="TRN"/></tracking>"""

FEED_LAST = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.15\nNS-2613,Sumatra,7.90\nNS-2614,Kenya,10.20\n"
FEED_NOW = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.80\nNS-2613,Sumatra,7.70\nNS-2616,Guatemala,8.75\n"

import time

def fetch_orders(shop, attempts=4, sleep=time.sleep):
    """Every order from every page, retrying transient failures with backoff."""
    orders, page = [], 1
    while True:
        for attempt in range(1, attempts + 1):
            try:
                body = shop.get_orders(page)
                break
            except (TimeoutError, ConnectionError):
                if attempt == attempts:
                    raise
                sleep(min(30, 2 ** (attempt - 1)))
        orders.extend(body["orders"])
        if page >= body["pages"]:
            return orders
        page += 1

waits = []
orders = fetch_orders(FakeShop(fail_first=2), sleep=waits.append)
print(len(orders), "orders fetched; waited", waits, "seconds after the failures")
print([o["id"] for o in orders])

Output

8 orders fetched; waited [1, 2] seconds after the failures
[5101, 5102, 5103, 5104, 5105, 5106, 5107, 5108]

Summarizing sales

summarize keeps only paid orders for the totals, counts refunds separately, and adds up bags per coffee with a Counter. Prices arrive as strings and become Decimals before any arithmetic, so the sales total is exact. Because the function receives a plain list of orders, testing it needs no network at all.

Totals and bags per coffee from the week's orders

from decimal import Decimal

# ---- fake sources for one week (2 to 8 November 2026) ----
ORDERS = [
    {"id": 5101, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10101", "lines": [["RC-101", 2, "18.50"], ["RC-104", 1, "20.25"]]},
    {"id": 5102, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10102", "lines": [["RC-102", 1, "22.00"]]},
    {"id": 5103, "placed": "2026-11-03", "status": "refunded", "tracking": None, "lines": [["RC-101", 1, "18.50"]]},
    {"id": 5104, "placed": "2026-11-04", "status": "paid", "tracking": "RDG10104", "lines": [["RC-106", 2, "21.50"], ["RC-101", 1, "18.50"]]},
    {"id": 5105, "placed": "2026-11-05", "status": "paid", "tracking": "RDG10105", "lines": [["RC-103", 3, "19.75"]]},
    {"id": 5106, "placed": "2026-11-06", "status": "paid", "tracking": "RDG10106", "lines": [["RC-101", 4, "18.50"]]},
    {"id": 5107, "placed": "2026-11-07", "status": "paid", "tracking": "RDG10107", "lines": [["RC-104", 2, "20.25"]]},
    {"id": 5108, "placed": "2026-11-08", "status": "paid", "tracking": None, "lines": [["RC-102", 1, "22.00"], ["RC-105", 1, "19.00"]]},
]

class FakeShop:
    """Pages of three orders, like the real /orders endpoint."""
    def __init__(self, fail_first=0):
        self.fail_first = fail_first
    def get_orders(self, page):
        if self.fail_first:
            self.fail_first -= 1
            raise TimeoutError("shop API timed out")
        pages = (len(ORDERS) + 2) // 3
        return {"page": page, "pages": pages, "orders": ORDERS[(page - 1) * 3: page * 3]}

TRACKING_XML = """<tracking><parcel number="RDG10101" code="DLV"/><parcel number="RDG10102" code="DLV"/>
<parcel number="RDG10104" code="DLV"/><parcel number="RDG10105" code="TRN"/>
<parcel number="RDG10106" code="EXC" note="address not found"/><parcel number="RDG10107" code="TRN"/></tracking>"""

FEED_LAST = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.15\nNS-2613,Sumatra,7.90\nNS-2614,Kenya,10.20\n"
FEED_NOW = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.80\nNS-2613,Sumatra,7.70\nNS-2616,Guatemala,8.75\n"

from collections import Counter

def summarize(orders):
    paid = [o for o in orders if o["status"] == "paid"]
    bags = Counter()
    revenue = Decimal("0")
    for order in paid:
        for sku, qty, price in order["lines"]:
            bags[sku] += qty
            revenue += Decimal(price) * qty
    refunds = sum(1 for o in orders if o["status"] == "refunded")
    return {"orders": len(paid), "refunds": refunds, "revenue": revenue, "bags": bags}

s = summarize(ORDERS)
print(s["orders"], "paid orders,", s["refunds"], "refund, sales", s["revenue"])
print(s["bags"].most_common())

Output

7 paid orders, 1 refund, sales 355.50
[('RC-101', 7), ('RC-104', 3), ('RC-103', 3), ('RC-102', 2), ('RC-106', 2), ('RC-105', 1)]

Try it yourself

The top coffees

Write top_coffees(bags, n=3) that takes a dictionary (or Counter) of bags sold per SKU and returns a list of the n best sellers as (sku, bags) tuples, most bags first. When two coffees sold the same number of bags, the one with the alphabetically smaller SKU comes first, so the report is the same every time.

def top_coffees(bags, n=3):
    return list(bags.items())[:n]

print(top_coffees({'RC-101': 7, 'RC-104': 3, 'RC-103': 3, 'RC-102': 2}))

Deliveries from the tracking feed

deliveries reads the carrier's XML once into a dictionary from tracking number to status code, then walks the paid orders. Orders without a tracking number are not yet shipped; parcels that are neither delivered nor in transit become action items with the carrier's note. A tracking number missing from the feed is reported too, rather than silently ignored, because it usually means a label was printed but never scanned.

Counting parcel statuses and finding the ones that need action

from decimal import Decimal

# ---- fake sources for one week (2 to 8 November 2026) ----
ORDERS = [
    {"id": 5101, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10101", "lines": [["RC-101", 2, "18.50"], ["RC-104", 1, "20.25"]]},
    {"id": 5102, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10102", "lines": [["RC-102", 1, "22.00"]]},
    {"id": 5103, "placed": "2026-11-03", "status": "refunded", "tracking": None, "lines": [["RC-101", 1, "18.50"]]},
    {"id": 5104, "placed": "2026-11-04", "status": "paid", "tracking": "RDG10104", "lines": [["RC-106", 2, "21.50"], ["RC-101", 1, "18.50"]]},
    {"id": 5105, "placed": "2026-11-05", "status": "paid", "tracking": "RDG10105", "lines": [["RC-103", 3, "19.75"]]},
    {"id": 5106, "placed": "2026-11-06", "status": "paid", "tracking": "RDG10106", "lines": [["RC-101", 4, "18.50"]]},
    {"id": 5107, "placed": "2026-11-07", "status": "paid", "tracking": "RDG10107", "lines": [["RC-104", 2, "20.25"]]},
    {"id": 5108, "placed": "2026-11-08", "status": "paid", "tracking": None, "lines": [["RC-102", 1, "22.00"], ["RC-105", 1, "19.00"]]},
]

class FakeShop:
    """Pages of three orders, like the real /orders endpoint."""
    def __init__(self, fail_first=0):
        self.fail_first = fail_first
    def get_orders(self, page):
        if self.fail_first:
            self.fail_first -= 1
            raise TimeoutError("shop API timed out")
        pages = (len(ORDERS) + 2) // 3
        return {"page": page, "pages": pages, "orders": ORDERS[(page - 1) * 3: page * 3]}

TRACKING_XML = """<tracking><parcel number="RDG10101" code="DLV"/><parcel number="RDG10102" code="DLV"/>
<parcel number="RDG10104" code="DLV"/><parcel number="RDG10105" code="TRN"/>
<parcel number="RDG10106" code="EXC" note="address not found"/><parcel number="RDG10107" code="TRN"/></tracking>"""

FEED_LAST = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.15\nNS-2613,Sumatra,7.90\nNS-2614,Kenya,10.20\n"
FEED_NOW = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.80\nNS-2613,Sumatra,7.70\nNS-2616,Guatemala,8.75\n"

from collections import Counter

def summarize(orders):
    paid = [o for o in orders if o["status"] == "paid"]
    bags = Counter()
    revenue = Decimal("0")
    for order in paid:
        for sku, qty, price in order["lines"]:
            bags[sku] += qty
            revenue += Decimal(price) * qty
    refunds = sum(1 for o in orders if o["status"] == "refunded")
    return {"orders": len(paid), "refunds": refunds, "revenue": revenue, "bags": bags}

import xml.etree.ElementTree as ET

NAMES = {"DLV": "delivered", "TRN": "in transit", "EXC": "exception"}

def deliveries(xml_text, orders):
    codes = {p.get("number"): (p.get("code"), p.get("note", "")) for p in ET.fromstring(xml_text).iter("parcel")}
    counts, problems, unshipped = Counter(), [], []
    for order in orders:
        if order["status"] != "paid":
            continue
        number = order["tracking"]
        if number is None:
            unshipped.append(order["id"])
            continue
        code, note = codes.get(number, ("UNK", "no tracking data"))
        counts[NAMES.get(code, "unknown")] += 1
        if code not in ("DLV", "TRN"):
            problems.append(f"order {order['id']} ({number}): {note}")
    return {"counts": counts, "problems": problems, "unshipped": unshipped}

d = deliveries(TRACKING_XML, ORDERS)
print(dict(d["counts"]))
print("action:", d["problems"])
print("not shipped:", d["unshipped"])

Output

{'delivered': 3, 'in transit': 2, 'exception': 1}
action: ['order 5106 (RDG10106): address not found']
not shipped: [5108]

Price changes from the importer's feed

price_changes reads last week's and this week's CSV feeds into dictionaries of Decimal prices and compares them with set operations: lots in both whose price moved by more than the limit, lots only in the new feed, and lots only in the old one. Kai only wants to hear about real changes, so the limit filters out small adjustments.

Comparing two weeks of the price feed

from decimal import Decimal

# ---- fake sources for one week (2 to 8 November 2026) ----
ORDERS = [
    {"id": 5101, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10101", "lines": [["RC-101", 2, "18.50"], ["RC-104", 1, "20.25"]]},
    {"id": 5102, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10102", "lines": [["RC-102", 1, "22.00"]]},
    {"id": 5103, "placed": "2026-11-03", "status": "refunded", "tracking": None, "lines": [["RC-101", 1, "18.50"]]},
    {"id": 5104, "placed": "2026-11-04", "status": "paid", "tracking": "RDG10104", "lines": [["RC-106", 2, "21.50"], ["RC-101", 1, "18.50"]]},
    {"id": 5105, "placed": "2026-11-05", "status": "paid", "tracking": "RDG10105", "lines": [["RC-103", 3, "19.75"]]},
    {"id": 5106, "placed": "2026-11-06", "status": "paid", "tracking": "RDG10106", "lines": [["RC-101", 4, "18.50"]]},
    {"id": 5107, "placed": "2026-11-07", "status": "paid", "tracking": "RDG10107", "lines": [["RC-104", 2, "20.25"]]},
    {"id": 5108, "placed": "2026-11-08", "status": "paid", "tracking": None, "lines": [["RC-102", 1, "22.00"], ["RC-105", 1, "19.00"]]},
]

class FakeShop:
    """Pages of three orders, like the real /orders endpoint."""
    def __init__(self, fail_first=0):
        self.fail_first = fail_first
    def get_orders(self, page):
        if self.fail_first:
            self.fail_first -= 1
            raise TimeoutError("shop API timed out")
        pages = (len(ORDERS) + 2) // 3
        return {"page": page, "pages": pages, "orders": ORDERS[(page - 1) * 3: page * 3]}

TRACKING_XML = """<tracking><parcel number="RDG10101" code="DLV"/><parcel number="RDG10102" code="DLV"/>
<parcel number="RDG10104" code="DLV"/><parcel number="RDG10105" code="TRN"/>
<parcel number="RDG10106" code="EXC" note="address not found"/><parcel number="RDG10107" code="TRN"/></tracking>"""

FEED_LAST = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.15\nNS-2613,Sumatra,7.90\nNS-2614,Kenya,10.20\n"
FEED_NOW = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.80\nNS-2613,Sumatra,7.70\nNS-2616,Guatemala,8.75\n"

import csv, io

def read_feed(text):
    return {r["lot"]: Decimal(r["price_per_kg"]) for r in csv.DictReader(io.StringIO(text))}

def price_changes(old_text, new_text, limit=Decimal("0.50")):
    old, new = read_feed(old_text), read_feed(new_text)
    moved = [(lot, old[lot], new[lot]) for lot in sorted(old.keys() & new.keys()) if abs(new[lot] - old[lot]) > limit]
    return {"moved": moved, "added": sorted(new.keys() - old.keys()), "removed": sorted(old.keys() - new.keys())}

changes = price_changes(FEED_LAST, FEED_NOW)
for lot, old, new in changes["moved"]:
    print(f"{lot}: {old} -> {new} ({new - old:+} per kg)")
print("new:", changes["added"], "| gone:", changes["removed"])

Output

NS-2612: 8.15 -> 8.80 (+0.65 per kg)
new: ['NS-2616'] | gone: ['NS-2614']

Try it yourself

Which prices moved?

Write moves(old, new, limit) where old and new are dictionaries mapping lot codes to Decimal prices and limit is a Decimal. Return a sorted list of (lot, change) tuples for lots present in both whose price changed by more than limit in either direction, where change is new - old.

def moves(old, new, limit):
    return [(lot, new[lot] - old[lot]) for lot in new if new[lot] - old[lot] > limit]

print(moves(OLD, NEW, D('0.50')))

Rendering the report

render turns the results into the text Theo reads. It starts with the headline numbers, then the top coffees, then the delivery picture, and finally the lines marked ACTION and PRICE, the items that need a decision this week. Keeping rendering separate from calculation means the same results can be rendered differently, as JSON for another program or as HTML for a richer email, without touching the analysis.

The rendered weekly report

from decimal import Decimal

# ---- fake sources for one week (2 to 8 November 2026) ----
ORDERS = [
    {"id": 5101, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10101", "lines": [["RC-101", 2, "18.50"], ["RC-104", 1, "20.25"]]},
    {"id": 5102, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10102", "lines": [["RC-102", 1, "22.00"]]},
    {"id": 5103, "placed": "2026-11-03", "status": "refunded", "tracking": None, "lines": [["RC-101", 1, "18.50"]]},
    {"id": 5104, "placed": "2026-11-04", "status": "paid", "tracking": "RDG10104", "lines": [["RC-106", 2, "21.50"], ["RC-101", 1, "18.50"]]},
    {"id": 5105, "placed": "2026-11-05", "status": "paid", "tracking": "RDG10105", "lines": [["RC-103", 3, "19.75"]]},
    {"id": 5106, "placed": "2026-11-06", "status": "paid", "tracking": "RDG10106", "lines": [["RC-101", 4, "18.50"]]},
    {"id": 5107, "placed": "2026-11-07", "status": "paid", "tracking": "RDG10107", "lines": [["RC-104", 2, "20.25"]]},
    {"id": 5108, "placed": "2026-11-08", "status": "paid", "tracking": None, "lines": [["RC-102", 1, "22.00"], ["RC-105", 1, "19.00"]]},
]

class FakeShop:
    """Pages of three orders, like the real /orders endpoint."""
    def __init__(self, fail_first=0):
        self.fail_first = fail_first
    def get_orders(self, page):
        if self.fail_first:
            self.fail_first -= 1
            raise TimeoutError("shop API timed out")
        pages = (len(ORDERS) + 2) // 3
        return {"page": page, "pages": pages, "orders": ORDERS[(page - 1) * 3: page * 3]}

TRACKING_XML = """<tracking><parcel number="RDG10101" code="DLV"/><parcel number="RDG10102" code="DLV"/>
<parcel number="RDG10104" code="DLV"/><parcel number="RDG10105" code="TRN"/>
<parcel number="RDG10106" code="EXC" note="address not found"/><parcel number="RDG10107" code="TRN"/></tracking>"""

FEED_LAST = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.15\nNS-2613,Sumatra,7.90\nNS-2614,Kenya,10.20\n"
FEED_NOW = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.80\nNS-2613,Sumatra,7.70\nNS-2616,Guatemala,8.75\n"

from collections import Counter

def summarize(orders):
    paid = [o for o in orders if o["status"] == "paid"]
    bags = Counter()
    revenue = Decimal("0")
    for order in paid:
        for sku, qty, price in order["lines"]:
            bags[sku] += qty
            revenue += Decimal(price) * qty
    refunds = sum(1 for o in orders if o["status"] == "refunded")
    return {"orders": len(paid), "refunds": refunds, "revenue": revenue, "bags": bags}

import xml.etree.ElementTree as ET

NAMES = {"DLV": "delivered", "TRN": "in transit", "EXC": "exception"}

def deliveries(xml_text, orders):
    codes = {p.get("number"): (p.get("code"), p.get("note", "")) for p in ET.fromstring(xml_text).iter("parcel")}
    counts, problems, unshipped = Counter(), [], []
    for order in orders:
        if order["status"] != "paid":
            continue
        number = order["tracking"]
        if number is None:
            unshipped.append(order["id"])
            continue
        code, note = codes.get(number, ("UNK", "no tracking data"))
        counts[NAMES.get(code, "unknown")] += 1
        if code not in ("DLV", "TRN"):
            problems.append(f"order {order['id']} ({number}): {note}")
    return {"counts": counts, "problems": problems, "unshipped": unshipped}

import csv, io

def read_feed(text):
    return {r["lot"]: Decimal(r["price_per_kg"]) for r in csv.DictReader(io.StringIO(text))}

def price_changes(old_text, new_text, limit=Decimal("0.50")):
    old, new = read_feed(old_text), read_feed(new_text)
    moved = [(lot, old[lot], new[lot]) for lot in sorted(old.keys() & new.keys()) if abs(new[lot] - old[lot]) > limit]
    return {"moved": moved, "added": sorted(new.keys() - old.keys()), "removed": sorted(old.keys() - new.keys())}

def render(week, summary, deliv, prices):
    out = [f"Ridgeline weekly report: {week}", "",
           f"Orders paid: {summary['orders']}   refunds: {summary['refunds']}   sales: {summary['revenue']:.2f}"]
    top = sorted(summary["bags"].items(), key=lambda kv: (-kv[1], kv[0]))[:3]
    out.append("Top coffees: " + ", ".join(f"{sku} ({n} bags)" for sku, n in top))
    c = deliv["counts"]
    out.append(f"Parcels: delivered {c['delivered']}, in transit {c['in transit']}, exceptions {c['exception']}; "
               f"paid orders not yet shipped: {len(deliv['unshipped'])}")
    for p in deliv["problems"]:
        out.append("  ACTION " + p)
    for lot, old, new in prices["moved"]:
        out.append(f"  PRICE {lot}: {old} -> {new} per kg")
    if prices["added"]:
        out.append("  New lots: " + ", ".join(prices["added"]))
    if prices["removed"]:
        out.append("  No longer offered: " + ", ".join(prices["removed"]))
    return "\n".join(out)

print(render("2026-11-02 to 2026-11-08", summarize(ORDERS), deliveries(TRACKING_XML, ORDERS),
             price_changes(FEED_LAST, FEED_NOW)))

Output

Ridgeline weekly report: 2026-11-02 to 2026-11-08

Orders paid: 7   refunds: 1   sales: 355.50
Top coffees: RC-101 (7 bags), RC-103 (3 bags), RC-104 (3 bags)
Parcels: delivered 3, in transit 2, exceptions 1; paid orders not yet shipped: 1
  ACTION order 5106 (RDG10106): address not found
  PRICE NS-2612: 8.15 -> 8.80 per kg
  New lots: NS-2616
  No longer offered: NS-2614

Try it yourself

The CSV attachment

Write bags_csv(bags) that returns the text of the CSV attachment: a header line sku,bags, then one line per coffee in SKU order, each ending with a newline. Use the csv module with lineterminator="\n" so the output is the same on every system.

def bags_csv(bags):
    return 'sku,bags\n' + ''.join(f'{k},{v}\n' for k, v in bags.items())

print(bags_csv({'RC-104': 3, 'RC-101': 7}))

The command-line tool

main ties everything together with the options from lesson 10. Without --email it prints the report, which is what people want when they run it by hand. With --email it builds the message with the CSV attachment and sends it, unless --dry-run is given or no sending function was provided. A source that cannot be reached after the retries turns into a clear message and exit code 1, so the scheduler and the monitoring notice. The listing runs the complete tool three ways: by hand, as the scheduler would but in dry-run mode, and against a shop that never answers.

The complete tool: JSON output, a dry-run email, and a shop that is down

from decimal import Decimal

# ---- fake sources for one week (2 to 8 November 2026) ----
ORDERS = [
    {"id": 5101, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10101", "lines": [["RC-101", 2, "18.50"], ["RC-104", 1, "20.25"]]},
    {"id": 5102, "placed": "2026-11-02", "status": "paid", "tracking": "RDG10102", "lines": [["RC-102", 1, "22.00"]]},
    {"id": 5103, "placed": "2026-11-03", "status": "refunded", "tracking": None, "lines": [["RC-101", 1, "18.50"]]},
    {"id": 5104, "placed": "2026-11-04", "status": "paid", "tracking": "RDG10104", "lines": [["RC-106", 2, "21.50"], ["RC-101", 1, "18.50"]]},
    {"id": 5105, "placed": "2026-11-05", "status": "paid", "tracking": "RDG10105", "lines": [["RC-103", 3, "19.75"]]},
    {"id": 5106, "placed": "2026-11-06", "status": "paid", "tracking": "RDG10106", "lines": [["RC-101", 4, "18.50"]]},
    {"id": 5107, "placed": "2026-11-07", "status": "paid", "tracking": "RDG10107", "lines": [["RC-104", 2, "20.25"]]},
    {"id": 5108, "placed": "2026-11-08", "status": "paid", "tracking": None, "lines": [["RC-102", 1, "22.00"], ["RC-105", 1, "19.00"]]},
]

class FakeShop:
    """Pages of three orders, like the real /orders endpoint."""
    def __init__(self, fail_first=0):
        self.fail_first = fail_first
    def get_orders(self, page):
        if self.fail_first:
            self.fail_first -= 1
            raise TimeoutError("shop API timed out")
        pages = (len(ORDERS) + 2) // 3
        return {"page": page, "pages": pages, "orders": ORDERS[(page - 1) * 3: page * 3]}

TRACKING_XML = """<tracking><parcel number="RDG10101" code="DLV"/><parcel number="RDG10102" code="DLV"/>
<parcel number="RDG10104" code="DLV"/><parcel number="RDG10105" code="TRN"/>
<parcel number="RDG10106" code="EXC" note="address not found"/><parcel number="RDG10107" code="TRN"/></tracking>"""

FEED_LAST = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.15\nNS-2613,Sumatra,7.90\nNS-2614,Kenya,10.20\n"
FEED_NOW = "lot,origin,price_per_kg\nNS-2611,Ethiopia,9.40\nNS-2612,Colombia,8.80\nNS-2613,Sumatra,7.70\nNS-2616,Guatemala,8.75\n"

import time

def fetch_orders(shop, attempts=4, sleep=time.sleep):
    """Every order from every page, retrying transient failures with backoff."""
    orders, page = [], 1
    while True:
        for attempt in range(1, attempts + 1):
            try:
                body = shop.get_orders(page)
                break
            except (TimeoutError, ConnectionError):
                if attempt == attempts:
                    raise
                sleep(min(30, 2 ** (attempt - 1)))
        orders.extend(body["orders"])
        if page >= body["pages"]:
            return orders
        page += 1

from collections import Counter

def summarize(orders):
    paid = [o for o in orders if o["status"] == "paid"]
    bags = Counter()
    revenue = Decimal("0")
    for order in paid:
        for sku, qty, price in order["lines"]:
            bags[sku] += qty
            revenue += Decimal(price) * qty
    refunds = sum(1 for o in orders if o["status"] == "refunded")
    return {"orders": len(paid), "refunds": refunds, "revenue": revenue, "bags": bags}

import xml.etree.ElementTree as ET

NAMES = {"DLV": "delivered", "TRN": "in transit", "EXC": "exception"}

def deliveries(xml_text, orders):
    codes = {p.get("number"): (p.get("code"), p.get("note", "")) for p in ET.fromstring(xml_text).iter("parcel")}
    counts, problems, unshipped = Counter(), [], []
    for order in orders:
        if order["status"] != "paid":
            continue
        number = order["tracking"]
        if number is None:
            unshipped.append(order["id"])
            continue
        code, note = codes.get(number, ("UNK", "no tracking data"))
        counts[NAMES.get(code, "unknown")] += 1
        if code not in ("DLV", "TRN"):
            problems.append(f"order {order['id']} ({number}): {note}")
    return {"counts": counts, "problems": problems, "unshipped": unshipped}

import csv, io

def read_feed(text):
    return {r["lot"]: Decimal(r["price_per_kg"]) for r in csv.DictReader(io.StringIO(text))}

def price_changes(old_text, new_text, limit=Decimal("0.50")):
    old, new = read_feed(old_text), read_feed(new_text)
    moved = [(lot, old[lot], new[lot]) for lot in sorted(old.keys() & new.keys()) if abs(new[lot] - old[lot]) > limit]
    return {"moved": moved, "added": sorted(new.keys() - old.keys()), "removed": sorted(old.keys() - new.keys())}

def render(week, summary, deliv, prices):
    out = [f"Ridgeline weekly report: {week}", "",
           f"Orders paid: {summary['orders']}   refunds: {summary['refunds']}   sales: {summary['revenue']:.2f}"]
    top = sorted(summary["bags"].items(), key=lambda kv: (-kv[1], kv[0]))[:3]
    out.append("Top coffees: " + ", ".join(f"{sku} ({n} bags)" for sku, n in top))
    c = deliv["counts"]
    out.append(f"Parcels: delivered {c['delivered']}, in transit {c['in transit']}, exceptions {c['exception']}; "
               f"paid orders not yet shipped: {len(deliv['unshipped'])}")
    for p in deliv["problems"]:
        out.append("  ACTION " + p)
    for lot, old, new in prices["moved"]:
        out.append(f"  PRICE {lot}: {old} -> {new} per kg")
    if prices["added"]:
        out.append("  New lots: " + ", ".join(prices["added"]))
    if prices["removed"]:
        out.append("  No longer offered: " + ", ".join(prices["removed"]))
    return "\n".join(out)

import argparse, json
from email.message import EmailMessage

def build_email(report, to, bags):
    msg = EmailMessage()
    msg["From"] = "Ridgeline Reports <reports@ridgelinecoffee.example>"
    msg["To"] = to
    msg["Subject"] = report.splitlines()[0]
    msg.set_content(report + "\n")
    rows = "sku,bags\n" + "".join(f"{sku},{n}\n" for sku, n in sorted(bags.items()))
    msg.add_attachment(rows.encode(), maintype="text", subtype="csv", filename="bags-sold.csv")
    return msg

def main(argv=None, shop=None, tracking=TRACKING_XML, feeds=(FEED_LAST, FEED_NOW), send=None, sleep=time.sleep):
    p = argparse.ArgumentParser(prog="weekly-report")
    p.add_argument("--week", default="2026-11-02 to 2026-11-08")
    p.add_argument("--format", choices=["text", "json"], default="text")
    p.add_argument("--email")
    p.add_argument("--dry-run", action="store_true")
    args = p.parse_args(argv)
    try:
        orders = fetch_orders(shop or FakeShop(), sleep=sleep)
    except (TimeoutError, ConnectionError) as e:
        print("could not reach the shop:", e)
        return 1
    summary = summarize(orders)
    deliv = deliveries(tracking, orders)
    prices = price_changes(*feeds)
    if args.format == "json":
        report = json.dumps({"week": args.week, "orders": summary["orders"], "revenue": str(summary["revenue"]),
                             "bags": dict(summary["bags"]), "problems": deliv["problems"]}, indent=2, sort_keys=True)
    else:
        report = render(args.week, summary, deliv, prices)
    if args.email:
        msg = build_email(report, args.email, summary["bags"])
        if args.dry_run or send is None:
            print(f"[dry run] would email {msg['To']}: {msg['Subject']}")
        else:
            send(msg)
    else:
        print(report)
    return 0

print("exit", main(["--format", "json"], sleep=lambda s: None))
print("exit", main(["--email", "theo@ridgelinecoffee.example", "--dry-run"], sleep=lambda s: None))
print("exit", main([], shop=FakeShop(fail_first=99), sleep=lambda s: None))

Output

{
  "bags": {
    "RC-101": 7,
    "RC-102": 2,
    "RC-103": 3,
    "RC-104": 3,
    "RC-105": 1,
    "RC-106": 2
  },
  "orders": 7,
  "problems": [
    "order 5106 (RDG10106): address not found"
  ],
  "revenue": "355.50",
  "week": "2026-11-02 to 2026-11-08"
}
exit 0
[dry run] would email theo@ridgelinecoffee.example: Ridgeline weekly report: 2026-11-02 to 2026-11-08
exit 0
could not reach the shop: shop API timed out
exit 1

On the shop server, a thin wrapper calls main with the real sources: a ShopClient built on a requests Session with the read-only key from the environment, the tracking XML fetched from the carrier with the cached OAuth token from lesson 3, the two feed files downloaded with Basic authentication, and a send function that uses smtplib with the app password. None of the analysis code changes. The last line of the wrapper is sys.exit(main()), so the exit code reaches the scheduler.

The production wrapper on the shop server (needs the real services)

import os, sys, smtplib, ssl
from ridgeline.reports import main
from ridgeline.sources import ShopClient, fetch_tracking_xml, fetch_feed

def send(msg):
    with smtplib.SMTP("smtp.mailprovider.example", 587, timeout=20) as s:
        s.starttls(context=ssl.create_default_context())
        s.login("reports@ridgelinecoffee.example", os.environ["RIDGELINE_MAIL_PASSWORD"])
        s.send_message(msg)

if __name__ == "__main__":
    shop = ShopClient.from_env()                         # reads RIDGELINE_SHOP_KEY
    code = main(shop=shop, tracking=fetch_tracking_xml(), feeds=(fetch_feed("last"), fetch_feed("this")), send=send)
    if code == 0:
        open("/var/lib/ridgeline/heartbeat/weekly-report", "w").write("ok\n")
    sys.exit(code)

Testing, scheduling and monitoring

The tests for the tool use the same fake week as this page. Each pure function gets a few tests with fixed inputs and exact expected outputs, including the awkward cases: a week with no orders, a parcel missing from the tracking feed, a lot whose price changed by exactly the limit, a refund. main gets tests for the three paths shown above, using the list form of argv and a send function that records messages instead of sending them. The suite runs in under a second and needs no network, so it runs on every commit in the continuous integration pipeline PYTH-260 will build.

Scheduling follows lesson 11. The crontab line runs the wrapper at 06:30 on Mondays, a time unaffected by daylight saving changes, under flock so a slow run cannot overlap a manual one. Every successful run writes the heartbeat file, and the monitoring check alerts the team chat if no heartbeat has been written for eight days. The runbook in docs/runbooks/weekly-report.md lists the sources, the credentials each needs, the log location, how to re-run the report by hand with --dry-run first, and who to call at the carrier and the importer when their feeds fail.

Worked problem

Reviewing the tool before it goes live

Before switching the schedule on, Jordan asks Sam to review the tool with a checklist built from this course. What should the review check?

Show solution

Credentials: every key and password comes from the environment; the shop key is read-only; nothing secret is printed or logged (lesson 3).
Network: every call has a timeout; only transient failures are retried, with backoff and a limit; the tool never sends a POST (lessons 1, 2, 11).
Data: money is Decimal from start to finish; feeds are validated; XML from the carrier is parsed with a known structure; the CSV attachment is written with the csv module (lessons 4, 7).
Behaviour: run by hand, it prints and never emails unless --email is given; failures give non-zero exit codes and clear messages (lessons 9, 10).
Operations: the schedule avoids daylight saving gaps, runs under a lock, writes a heartbeat, and has a runbook (lesson 11).

Answer: Sam found one problem: a debug print of the shop response that included the Authorization header. It was removed before the first scheduled run.

Common mistakes

MistakeStarting to code before agreeing what the report must contain.
InsteadWrite short, checkable requirements with the people who will use it.
MistakeMixing network calls into the calculation functions.
InsteadFetch first, then pass plain data to pure functions that are easy to test.
MistakeTesting the tool only against the live services.
InsteadTest every piece against fixed fake data, and keep the real services for a final check.
MistakeLetting a failed source produce a report with silently missing numbers.
InsteadFail with a clear message and a non-zero exit code, or mark the section as unavailable.
MistakeEmailing the report whenever the script runs.
InsteadSend only when asked with --email, and offer --dry-run.
MistakeSwitching on the schedule without monitoring.
InsteadWrite a heartbeat on success, alert when it goes stale, and keep a runbook.

Summary and key terms

The capstone combines every lesson: APIs, authentication, data formats, documents, email, CLIs and reliable scheduling.
Requirements written as short, checkable statements drive the design and the tests.
Network access is isolated in fetch functions; summarizing, analysis and rendering are pure functions on plain data.
Sources and the send function are parameters, so the same code runs against fakes in tests and real services in production.
Money stays Decimal, feeds are validated and attachments are written with the csv module.
The command-line main offers text or JSON output, optional email with a dry run, and meaningful exit codes.
A production wrapper supplies real sources; scheduling, locking, heartbeats and a runbook make it safe to leave unattended.

requirement: A short, checkable statement of what the software must do.
pure function: A function whose result depends only on its inputs and that changes nothing outside itself.
data flow: The path data takes from sources through processing to outputs.
fake source: A stand-in with the same interface as a real service, used for testing.
wrapper: A thin script that connects tested code to real resources.
action item: A line in a report that needs someone to do something.
dry run: A run that shows what would happen without doing it.
heartbeat file: A file written after each successful run, checked by monitoring.
runbook: Instructions for operating a job and dealing with its failures.
code review: A check of code by another person before it is used.

Practice problems

Work each problem on paper first, then reveal the answer to check.

Why are summarize, deliveries and price_changes easy to test?
Show answer
They take plain data and return plain data, with no network or files involved.
What does the tool do when the shop API times out every time?
Show answer
It gives up after the retries, prints a clear message and returns exit code 1.
Why is the shop key for the report read-only?
Show answer
The report only reads data, so a leaked key could not change anything.
How does the tool avoid emailing Theo when Jordan runs it by hand?
Show answer
It only emails when --email is given, and --dry-run shows the message without sending.
Name two awkward cases the tests should cover.
Show answer
A week with no orders, a parcel missing from the tracking feed, a price change exactly equal to the limit, or a refund.
What alerts the team if the scheduled job stops running altogether?
Show answer
The monitoring check that finds the heartbeat file older than eight days.