Anti-patterns: god objects and golden hammers

Python · orders/measure.py

"""Measures a class with Python's ast module: its size, a proposed split of its fields, and its change history."""
import ast
from collections import Counter
from pathlib import Path

SOURCE, CLASS, HISTORY = "order_manager.py", "OrderManager", "history.txt"

# A proposed split: which fields belong together. Edit it and run the script again to test another split.
GROUPS = {
    "pricing": ["products", "coupons", "tax_rate_bps", "free_delivery_from", "delivery_fee", "express_fee"],
    "stock": ["stock", "held", "low_stock_level"],
    "orders": ["orders", "next_order_id"],
    "messages": ["outbox", "sms_enabled", "shop_name"],
    "invoices": ["invoices", "next_invoice_no", "sales_by_day"],
}

tree = ast.parse(Path(SOURCE).read_text())
cls = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == CLASS)
methods = [n for n in cls.body if isinstance(n, ast.FunctionDef)]
init = next(m for m in methods if m.name == "__init__")
fields = [t.attr for n in ast.walk(init) if isinstance(n, ast.Assign) for t in n.targets
          if isinstance(t, ast.Attribute) and isinstance(t.value, ast.Name) and t.value.id == "self"]
modules = sorted(a.name for n in tree.body if isinstance(n, ast.Import) for a in n.names)
group_of = {f: g for g, members in GROUPS.items() for f in members}
missing = [f for f in fields if f not in group_of]
assert not missing, f"put these fields in a group: {missing}"


def length(node):
    return node.end_lineno - node.lineno + 1


def self_attributes(node):
    return {n.attr for n in ast.walk(node)
            if isinstance(n, ast.Attribute) and isinstance(n.value, ast.Name) and n.value.id == "self"}


longest = max(methods, key=length)
print(f"{CLASS}: {length(cls)} lines, {len(methods)} methods, {len(fields)} fields, "
      f"imports {', '.join(modules)}")
print(f"Longest method: {longest.name}, {length(longest)} lines\n")

# Which groups each method reaches through the fields it reads or writes.
belongs, spans = {g: [] for g in GROUPS}, []
for m in methods:
    if m.name == "__init__":
        continue
    groups = sorted({group_of[a] for a in self_attributes(m) if a in group_of}, key=list(GROUPS).index)
    if len(groups) == 1:
        belongs[groups[0]].append(m)
    else:
        spans.append((m, groups))
print("The proposed split, method by method:")
for g, ms in belongs.items():
    print(f"  {g:<9}{len(GROUPS[g])} fields, {len(ms):>2} methods, {sum(map(length, ms)):>3} lines")
print(f"  {len(spans)} methods use fields of several groups:")
for m, groups in sorted(spans, key=lambda s: -len(s[1])):
    print(f"    {m.name:<20}{len(groups)} groups: {', '.join(groups)}")

# The history: one commit per line, "hash | message | the methods it changed".
commits = [line.split(" | ") for line in Path(HISTORY).read_text().splitlines() if line.strip()]
by_name = {m.name: m for m in methods}
touched = [[name.strip() for name in changed.split(",")] for _, _, changed in commits]
per_method = Counter(name for names in touched for name in names if name in by_name)
per_group = Counter(g for names in touched
                    for g in {next((gr for gr, ms in belongs.items() if by_name[n] in ms), "a cross-group method")
                              for n in names if n in by_name})
print(f"\nChange history: {len(commits)} commits, {sum(1 for t in touched if any(n in by_name for n in t))} "
      f"of them changed {CLASS}")
print("  commits that changed each part: " + ", ".join(f"{g} {c}" for g, c in per_group.most_common()))
print("  hotspots (changes x lines):")
for name, count in sorted(per_method.items(), key=lambda kv: (-kv[1] * length(by_name[kv[0]]), kv[0]))[:5]:
    print(f"    {name:<20}{count:>2} x {length(by_name[name]):>2} = {count * length(by_name[name])}")
quiet = max((m for m in methods if per_method[m.name] <= 1 and m.name != "__init__"), key=length)
print(f"  largest method changed only once: {quiet.name}, {length(quiet)} lines")

print("\nAfter the split, in order_service.py:")
for c in (n for n in ast.parse(Path("order_service.py").read_text()).body if isinstance(n, ast.ClassDef)):
    ms = [n for n in c.body if isinstance(n, ast.FunctionDef)]
    top = max((m for m in ms if m.name != "__init__"), key=length)
    print(f"  {c.name:<13}{length(c):>4} lines, {len(ms):>2} methods, longest {top.name} ({length(top)} lines)")

OrderManager: 250 lines, 30 methods, 17 fields, imports json, re
Longest method: place_order, 39 lines

The proposed split, method by method:
  pricing  6 fields,  8 methods,  55 lines
  stock    3 fields,  2 methods,   4 lines
  orders   2 fields,  4 methods,  10 lines
  messages 3 fields,  3 methods,   8 lines
  invoices 3 fields,  3 methods,  10 lines
  9 methods use fields of several groups:
    place_order         5 groups: pricing, stock, orders, messages, invoices
    cancel_order        4 groups: stock, orders, messages, invoices
    ship_order          3 groups: stock, orders, messages
    resend_invoice      3 groups: orders, messages, invoices
    remove_product      2 groups: pricing, stock
    restock             2 groups: pricing, stock
    low_stock_report    2 groups: pricing, stock
    best_sellers        2 groups: pricing, orders
    export_report_json  2 groups: orders, invoices

Change history: 60 commits, 56 of them changed OrderManager
  commits that changed each part: a cross-group method 36, pricing 23, invoices 5, orders 4, messages 3, stock 2
  hotspots (changes x lines):
    place_order         18 x 39 = 702
    quote               17 x 22 = 374
    cancel_order        10 x 19 = 190
    _invoice_text        4 x 14 = 56
    ship_order           5 x 10 = 50
  largest method changed only once: export_report_json, 10 lines

After the split, in order_service.py:
  Pricing        66 lines, 12 methods, longest quote (13 lines)
  Stock          38 lines, 10 methods, longest hold (4 lines)
  Messages       25 lines,  6 methods, longest clear (4 lines)
  Invoices       34 lines,  6 methods, longest issue (14 lines)
  OrderService  166 lines, 29 methods, longest place_order (29 lines)

orders/order_manager.py

"""A small online shop's whole order process in one class: an invented example of a god object.

Amounts are whole paise (100 paise make a rupee), so no rounding of floats creeps in.
"""
import json
import re


class OrderManager:
    """Prices, stock, orders, messages, invoices and reports, all in one place."""

    def __init__(self, tax_rate_bps=1800, free_delivery_from=49_900, express_fee=9_900):
        # pricing
        self.products = {}  # sku -> {"name": ..., "price": paise}
        self.coupons = {}  # code -> {"kind": "percent" or "flat", "value": ..., "min_spend": paise}
        self.tax_rate_bps = tax_rate_bps  # 1800 basis points is 18 %
        self.free_delivery_from = free_delivery_from
        self.delivery_fee = 4_000
        self.express_fee = express_fee
        # stock
        self.stock = {}  # sku -> units on the shelf
        self.held = {}  # order id -> [(sku, units)]
        self.low_stock_level = 3
        # orders
        self.orders = {}  # order id -> dict
        self.next_order_id = 1001
        # messages
        self.outbox = []  # (channel, to, text)
        self.sms_enabled = True
        self.shop_name = "Paper Plane"
        # invoices and reports
        self.invoices = {}  # invoice number -> text
        self.next_invoice_no = 1
        self.sales_by_day = {}  # day number -> paise

    # ---- catalogue ----------------------------------------------------------------------------
    def add_product(self, sku, name, price):
        if price <= 0:
            raise ValueError(f"price of {sku} must be positive")
        self.products[sku] = {"name": name, "price": price}

    def set_price(self, sku, price):
        if sku not in self.products:
            raise KeyError(sku)
        self.products[sku]["price"] = price

    def remove_product(self, sku):
        self.products.pop(sku, None)
        self.stock.pop(sku, None)

    def price_of(self, sku):
        return self.products[sku]["price"]

    def add_coupon(self, code, kind, value, min_spend=0):
        if kind not in ("percent", "flat"):
            raise ValueError(f"unknown coupon kind {kind}")
        self.coupons[code.upper()] = {"kind": kind, "value": value, "min_spend": min_spend}

    def expire_coupon(self, code):
        self.coupons.pop(code.upper(), None)

    def coupon_is_valid(self, code, subtotal):
        coupon = self.coupons.get(code.upper())
        return coupon is not None and subtotal >= coupon["min_spend"]

    def quote(self, items, coupon=None, express=False):
        subtotal = 0
        for sku, units in items:
            subtotal += self.products[sku]["price"] * units
        discount = 0
        if coupon:
            c = self.coupons.get(coupon.upper())
            if c and subtotal >= c["min_spend"]:
                if c["kind"] == "percent":
                    discount = subtotal * c["value"] // 100
                else:
                    discount = min(c["value"], subtotal)
        taxed = subtotal - discount
        tax = (taxed * self.tax_rate_bps + 5_000) // 10_000
        if express:
            delivery = self.express_fee
        elif taxed >= self.free_delivery_from:
            delivery = 0
        else:
            delivery = self.delivery_fee
        return {"subtotal": subtotal, "discount": discount, "tax": tax, "delivery": delivery,
                "total": taxed + tax + delivery}

    # ---- stock --------------------------------------------------------------------------------
    def restock(self, sku, units):
        if sku not in self.products:
            raise KeyError(sku)
        self.stock[sku] = self.stock.get(sku, 0) + units

    def available(self, sku):
        return self.stock.get(sku, 0)

    def units_held(self):
        return sum(units for lines in self.held.values() for _, units in lines)

    def low_stock_report(self):
        lines = []
        for sku in sorted(self.stock):
            if self.stock[sku] <= self.low_stock_level:
                lines.append(f"{sku} {self.products[sku]['name']}: {self.stock[sku]} left")
        return lines

    # ---- orders -------------------------------------------------------------------------------
    def place_order(self, customer, email, phone, items, coupon=None, express=False, day=1):
        if not items:
            raise ValueError("an order needs at least one item")
        if not re.fullmatch(r"\+?\d{10,13}", phone):
            raise ValueError(f"bad phone number {phone}")
        for sku, units in items:
            if sku not in self.products:
                raise ValueError(f"unknown product {sku}")
            if units <= 0:
                raise ValueError(f"bad quantity {units} for {sku}")
            if self.stock.get(sku, 0) < units:
                raise ValueError(f"only {self.stock.get(sku, 0)} of {sku} left")
        order_id = self.next_order_id
        self.next_order_id += 1
        for sku, units in items:
            self.stock[sku] -= units
        self.held[order_id] = list(items)
        prices = self.quote(items, coupon, express)
        self.orders[order_id] = {
            "id": order_id, "customer": customer, "email": email, "phone": phone,
            "items": list(items), "coupon": coupon, "express": express, "day": day,
            "status": "placed", "prices": prices, "tracking": None,
        }
        self.sales_by_day[day] = self.sales_by_day.get(day, 0) + prices["total"]
        total = prices["total"]
        rupees = f"{total // 100:,}.{total % 100:02d}"
        if self.sms_enabled:
            self.outbox.append(("sms", phone, f"{self.shop_name}: order {order_id} placed, {rupees} to pay."))
        body = f"Hello {customer},\nwe have your order {order_id}.\n"
        for sku, units in items:
            line = self.products[sku]["price"] * units
            body += f"  {units} x {self.products[sku]['name']}: {line // 100:,}.{line % 100:02d}\n"
        body += f"Total {rupees}" + (" (express delivery)" if express else "") + "\n"
        self.outbox.append(("email", email, f"Your order {order_id}\n{body}"))
        number = self.next_invoice_no
        self.next_invoice_no += 1
        self.invoices[number] = self._invoice_text(number, self.orders[order_id])
        self.orders[order_id]["invoice"] = number
        return order_id

    def cancel_order(self, order_id, reason, day=1):
        order = self.orders.get(order_id)
        if order is None:
            raise KeyError(order_id)
        if order["status"] == "shipped":
            raise ValueError(f"order {order_id} has shipped and cannot be cancelled")
        if order["status"] == "cancelled":
            return False
        for sku, units in self.held.pop(order_id, []):
            self.stock[sku] = self.stock.get(sku, 0) + units
        order["status"] = "cancelled"
        order["reason"] = reason
        self.sales_by_day[day] = self.sales_by_day.get(day, 0) - order["prices"]["total"]
        total = order["prices"]["total"]
        rupees = f"{total // 100:,}.{total % 100:02d}"
        if self.sms_enabled:
            self.outbox.append(("sms", order["phone"], f"{self.shop_name}: order {order_id} cancelled, {rupees} refunded."))
        self.outbox.append(("email", order["email"], f"Order {order_id} cancelled\nReason: {reason}\nRefund {rupees}\n"))
        return True

    def ship_order(self, order_id, tracking):
        order = self.orders[order_id]
        if order["status"] != "placed":
            raise ValueError(f"order {order_id} is {order['status']}")
        order["status"] = "shipped"
        order["tracking"] = tracking
        self.held.pop(order_id, None)
        if self.sms_enabled:
            self.outbox.append(("sms", order["phone"], f"{self.shop_name}: order {order_id} shipped, tracking {tracking}."))
        self.outbox.append(("email", order["email"], f"Order {order_id} is on its way\nTracking: {tracking}\n"))

    def status_of(self, order_id):
        return self.orders[order_id]["status"]

    def orders_of(self, customer):
        return [o["id"] for o in self.orders.values() if o["customer"] == customer]

    def customer_spend(self, customer):
        return sum(o["prices"]["total"] for o in self.orders.values()
                   if o["customer"] == customer and o["status"] != "cancelled")

    def refund_amount(self, order_id):
        order = self.orders[order_id]
        return order["prices"]["total"] if order["status"] == "cancelled" else 0

    # ---- messages -----------------------------------------------------------------------------
    def set_sms(self, enabled):
        self.sms_enabled = enabled

    def messages_for(self, to):
        return [text for channel, address, text in self.outbox if address == to]

    def clear_outbox(self):
        sent = len(self.outbox)
        self.outbox = []
        return sent

    def resend_invoice(self, order_id):
        order = self.orders[order_id]
        self.outbox.append(("email", order["email"], self.invoices[order["invoice"]]))

    # ---- invoices and reports -----------------------------------------------------------------
    def _invoice_text(self, number, order):
        p = order["prices"]
        def money(paise):
            return f"{paise // 100:,}.{paise % 100:02d}"
        rows = [f"Invoice {number} for order {order['id']}, {order['customer']}"]
        for sku, units in order["items"]:
            rows.append(f"  {sku:<12}{units:>3} x {money(self.products[sku]['price']):>10}")
        rows.append(f"  Subtotal {money(p['subtotal']):>18}")
        if p["discount"]:
            rows.append(f"  Discount {money(-p['discount']):>18}")
        rows.append(f"  Tax      {money(p['tax']):>18}")
        rows.append(f"  Delivery {money(p['delivery']):>18}")
        rows.append(f"  Total    {money(p['total']):>18}")
        return "\n".join(rows)

    def invoice_text(self, number):
        return self.invoices[number]

    def sales_report(self):
        lines = []
        for day in sorted(self.sales_by_day):
            paise = self.sales_by_day[day]
            lines.append(f"day {day}: {paise // 100:,}.{paise % 100:02d}")
        return lines

    def day_total(self, day):
        return self.sales_by_day.get(day, 0)

    def best_sellers(self, n=3):
        units = {}
        for order in self.orders.values():
            if order["status"] == "cancelled":
                continue
            for sku, count in order["items"]:
                units[sku] = units.get(sku, 0) + count
        ranked = sorted(units.items(), key=lambda kv: (-kv[1], kv[0]))
        return [f"{self.products[sku]['name']}: {count}" for sku, count in ranked[:n]]

    def export_report_json(self):
        report = {
            "orders": len(self.orders),
            "cancelled": sum(1 for o in self.orders.values() if o["status"] == "cancelled"),
            "shipped": sum(1 for o in self.orders.values() if o["status"] == "shipped"),
            "units_held": self.units_held(),
            "sales": {str(day): paise for day, paise in sorted(self.sales_by_day.items())},
            "low_stock": self.low_stock_report(),
        }
        return json.dumps(report, sort_keys=True)

orders/history.txt

3f2a9c1 | first version of the shop | add_product, price_of, quote, restock, available, place_order, status_of
8b41d07 | send an SMS when an order is placed | place_order, set_sms
c07e5aa | percent coupons | add_coupon, quote, place_order
1d9e3b2 | flat coupons and a minimum spend | add_coupon, coupon_is_valid, quote
e5c8f40 | cancel orders and refund | cancel_order, refund_amount
42aa1f9 | fix: refund sent twice when cancelling twice | cancel_order
9e06b3d | invoices | place_order, _invoice_text, invoice_text
71f0c2e | free delivery above 499 rupees | quote
b38d5a4 | shipping with tracking numbers | ship_order, status_of
0c2e7f1 | update the readme | (other files)
5a7b9e0 | tax rounding to the nearest paisa | quote
f61c3d8 | fix: coupon codes are case-sensitive | add_coupon, quote, coupon_is_valid
2e9a40b | low stock report | low_stock_report
aa03c6f | express delivery | quote, place_order
6d5e1b9 | show express delivery in the order email | place_order
97c2f3e | customer history page | orders_of, customer_spend
c8b6a12 | daily sales report | sales_report, place_order
3b7d0e4 | fix: cancelled orders counted in sales | cancel_order, sales_report
e1f4a83 | resend an invoice from the help desk | resend_invoice
54c9b7f | lint configuration | (other files)
0f8e2d6 | price changes from the catalogue screen | set_price
d2a6c40 | discount line on invoices | _invoice_text
8c1f5e9 | fix: negative discount on flat coupons above the subtotal | quote
19b0d7a | shop name in every SMS | place_order, cancel_order, ship_order
b4e7c31 | best sellers on the home page | best_sellers
7a3d9f2 | fix: best sellers counted cancelled orders | best_sellers
e0c5b18 | report as JSON for the dashboard | export_report_json
63f8a2d | wording of the cancellation email | cancel_order
d9b1e07 | coupon expiry | expire_coupon
2c4f6a3 | phone number check | place_order
f7a0c95 | fix: phone numbers with a leading plus sign | place_order
4e2b8d1 | tax rate from the settings screen | quote
a1d6f08 | remove discontinued products | remove_product
8f3c2b7 | upgrade the test runner | (other files)
5d0e9a6 | wording of the order email | place_order
c3a7e52 | holds released when an order ships | ship_order, units_held
91e4b0c | fix: stock not returned on cancel | cancel_order
0b6f3d8 | delivery fee raised to 40 rupees | quote
e8a2c17 | invoice layout for the accountant | _invoice_text
36d9f4a | festival coupon: 15 percent off | quote
fa5c0b2 | fix: express fee charged on free delivery | quote
72b1e9d | messages screen for the help desk | messages_for, clear_outbox
cd4e8a5 | SMS can be switched off | set_sms, place_order, cancel_order, ship_order
1e7f3c9 | refund line in the cancellation SMS | cancel_order
b9c0d4e | invoice numbers start again at 1 for the audit | place_order
4a8e2f7 | fix: rounding of the percent discount | quote
e6d3b10 | show prices with thousands separators | place_order, cancel_order, _invoice_text, sales_report
2f9c7a8 | update dependencies | (other files)
8d5b1e3 | minimum spend for the festival coupon | quote, coupon_is_valid
c1a4f69 | cancellation reason in the email | cancel_order
5e0d8b2 | restock checks the product exists | restock
a7f2c3d | free delivery threshold back to 499 | quote
3c6e9b1 | express fee lowered to 99 rupees | quote, place_order
d0b8f47 | fix: email total ignored the coupon | place_order
94e1a6c | day totals for the finance team | day_total
6b3f0d9 | fix: orders with no items | place_order
e2c7a5b | tracking number in the shipping SMS | ship_order
7f9d2e0 | quantity check on each line | place_order
b5a0e3f | flat coupons capped at the subtotal | quote
0e4c8b6 | fix: low stock report crashed on removed products | low_stock_report, remove_product

Python · orders/split_check.py

"""Characterization check: the same seeded scenarios run through the god object and the split, compared result by result."""
import random

from order_manager import OrderManager
from order_service import OrderService, Pricing

PRODUCTS = [("PEN-BLUE", "Blue gel pen", 2_500), ("NOTE-A5", "A5 notebook", 12_000), ("INK-BLK", "Black ink", 18_000),
            ("BAG-CNV", "Canvas bag", 34_900), ("DESK-LMP", "Desk lamp", 129_900)]
PEOPLE = [("Asha", "asha@example.com", "+919800000001"), ("Ben", "ben@example.com", "07700900002"),
          ("Chen", "chen@example.com", "+8613800000003"), ("Dee", "dee@example.com", "12-34")]


def scenario(shop, seed):
    """Runs one seeded day of business and returns everything the shop answered, in order."""
    rnd = random.Random(seed)
    results = []

    def call(name, *args):
        try:
            results.append((name, args, repr(getattr(shop, name)(*args))))
        except (KeyError, ValueError) as error:
            results.append((name, args, f"raises {type(error).__name__}: {error}"))

    for sku, name, price in PRODUCTS:
        call("add_product", sku, name, price)
        call("restock", sku, rnd.randint(1, 8))
    call("add_coupon", "save10", "percent", 10, 50_000)
    call("add_coupon", "FLAT200", "flat", 20_000)
    placed = []
    for step in range(14):
        action = rnd.choice(["place", "place", "place", "cancel", "ship", "restock", "price", "sms"])
        if action == "place":
            customer, email, phone = rnd.choice(PEOPLE)
            skus = rnd.sample([p[0] for p in PRODUCTS] + ["GHOST-1"], rnd.randint(1, 3))
            items = [(sku, rnd.randint(0, 4)) for sku in skus]
            coupon = rnd.choice([None, None, "SAVE10", "save10", "FLAT200", "OLD5"])
            express = rnd.random() < 0.25
            call("quote", items, coupon, express)
            before = len(results)
            call("place_order", customer, email, phone, items, coupon, express, 1 + step // 5)
            if not results[before][2].startswith("raises"):
                placed.append(int(results[before][2]))
        elif action == "cancel" and placed:
            call("cancel_order", rnd.choice(placed + [9999]), "changed my mind", 1 + step // 5)
        elif action == "ship" and placed:
            call("ship_order", rnd.choice(placed), f"TRK{seed:03d}{step:02d}")
        elif action == "restock":
            call("restock", rnd.choice(PRODUCTS)[0], rnd.randint(1, 5))
        elif action == "price":
            call("set_price", rnd.choice(PRODUCTS)[0], rnd.randint(10, 200) * 100)
        elif action == "sms":
            call("set_sms", rnd.random() < 0.5)
    for order_id in placed:
        call("status_of", order_id)
        call("refund_amount", order_id)
        call("resend_invoice", order_id)
    for customer, email, phone in PEOPLE:
        call("orders_of", customer)
        call("customer_spend", customer)
        call("messages_for", email)
        call("messages_for", phone)
    for number in range(1, len(placed) + 1):
        call("invoice_text", number)
    for name in ("low_stock_report", "sales_report", "best_sellers", "units_held", "export_report_json"):
        call(name)
    return results


def compare(make_new, label, seeds=range(40)):
    differing, values, first = 0, 0, None
    for seed in seeds:
        old, new = scenario(OrderManager(), seed), scenario(make_new(), seed)
        values += len(old)
        if old != new:
            differing += 1
            if first is None:
                i = next(i for i, (a, b) in enumerate(zip(old, new)) if a != b)
                first = (seed, old[i], new[i])
    print(f"{label}: {differing} of {len(seeds)} scenarios differ ({values} results compared)")
    if first:
        seed, (name, args, was), (_, _, now) = first
        print(f"  first difference, scenario {seed}: {name}{args}")
        print(f"    god object: {was}")
        print(f"    split:      {now}")


class ForgotMinimumSpend(Pricing):
    """The kind of slip a split invites: the coupon's minimum spend was left behind in the old class."""

    def discount(self, code, subtotal):
        coupon = self.coupons.get(code.upper()) if code else None
        if not coupon:
            return 0
        return subtotal * coupon["value"] // 100 if coupon["kind"] == "percent" else min(coupon["value"], subtotal)


compare(OrderService, "OrderService")
compare(lambda: OrderService(pricing=ForgotMinimumSpend()), "OrderService with one slip")

OrderService: 0 of 40 scenarios differ (2210 results compared)
OrderService with one slip: 17 of 40 scenarios differ (2210 results compared)
  first difference, scenario 0: quote([('PEN-BLUE', 4)], 'SAVE10', False)
    god object: {'subtotal': 10000, 'discount': 0, 'tax': 1800, 'delivery': 4000, 'total': 15800}
    split:      {'subtotal': 10000, 'discount': 1000, 'tax': 1620, 'delivery': 4000, 'total': 14620}

orders/order_service.py

"""The same shop after the split: four classes with one job each, and a facade that keeps the old public methods."""
import json
import re


def money(paise):
    return f"{paise // 100:,}.{paise % 100:02d}"


class Pricing:
    """The catalogue's prices and the coupons, tax and delivery rules."""

    def __init__(self, tax_rate_bps=1800, free_delivery_from=49_900, express_fee=9_900):
        self.products = {}
        self.coupons = {}
        self.tax_rate_bps = tax_rate_bps
        self.free_delivery_from = free_delivery_from
        self.delivery_fee = 4_000
        self.express_fee = express_fee

    def add_product(self, sku, name, price):
        if price <= 0:
            raise ValueError(f"price of {sku} must be positive")
        self.products[sku] = {"name": name, "price": price}

    def set_price(self, sku, price):
        if sku not in self.products:
            raise KeyError(sku)
        self.products[sku]["price"] = price

    def remove_product(self, sku):
        self.products.pop(sku, None)

    def has(self, sku):
        return sku in self.products

    def name_of(self, sku):
        return self.products[sku]["name"]

    def price_of(self, sku):
        return self.products[sku]["price"]

    def add_coupon(self, code, kind, value, min_spend=0):
        if kind not in ("percent", "flat"):
            raise ValueError(f"unknown coupon kind {kind}")
        self.coupons[code.upper()] = {"kind": kind, "value": value, "min_spend": min_spend}

    def expire_coupon(self, code):
        self.coupons.pop(code.upper(), None)

    def coupon_is_valid(self, code, subtotal):
        coupon = self.coupons.get(code.upper())
        return coupon is not None and subtotal >= coupon["min_spend"]

    def discount(self, code, subtotal):
        coupon = self.coupons.get(code.upper()) if code else None
        if not coupon or subtotal < coupon["min_spend"]:
            return 0
        if coupon["kind"] == "percent":
            return subtotal * coupon["value"] // 100
        return min(coupon["value"], subtotal)

    def quote(self, items, coupon=None, express=False):
        subtotal = sum(self.price_of(sku) * units for sku, units in items)
        discount = self.discount(coupon, subtotal)
        taxed = subtotal - discount
        tax = (taxed * self.tax_rate_bps + 5_000) // 10_000
        if express:
            delivery = self.express_fee
        elif taxed >= self.free_delivery_from:
            delivery = 0
        else:
            delivery = self.delivery_fee
        return {"subtotal": subtotal, "discount": discount, "tax": tax, "delivery": delivery,
                "total": taxed + tax + delivery}


class Stock:
    """Units on the shelf, and the units held for orders that have not shipped."""

    def __init__(self, low_stock_level=3):
        self.units = {}
        self.held = {}
        self.low_stock_level = low_stock_level

    def restock(self, sku, units):
        self.units[sku] = self.units.get(sku, 0) + units

    def available(self, sku):
        return self.units.get(sku, 0)

    def forget(self, sku):
        self.units.pop(sku, None)

    def check(self, sku, units):
        if self.available(sku) < units:
            raise ValueError(f"only {self.available(sku)} of {sku} left")

    def hold(self, order_id, items):
        for sku, units in items:
            self.units[sku] -= units
        self.held[order_id] = list(items)

    def release(self, order_id):
        for sku, units in self.held.pop(order_id, []):
            self.units[sku] = self.units.get(sku, 0) + units

    def settle(self, order_id):
        self.held.pop(order_id, None)

    def units_held(self):
        return sum(units for lines in self.held.values() for _, units in lines)

    def low(self):
        return [(sku, self.units[sku]) for sku in sorted(self.units) if self.units[sku] <= self.low_stock_level]


class Messages:
    """The outbox of text messages and emails."""

    def __init__(self, shop_name="Paper Plane"):
        self.outbox = []
        self.sms_enabled = True
        self.shop_name = shop_name

    def sms(self, phone, text):
        if self.sms_enabled:
            self.outbox.append(("sms", phone, f"{self.shop_name}: {text}"))

    def email(self, address, text):
        self.outbox.append(("email", address, text))

    def set_sms(self, enabled):
        self.sms_enabled = enabled

    def messages_for(self, to):
        return [text for channel, address, text in self.outbox if address == to]

    def clear(self):
        sent = len(self.outbox)
        self.outbox = []
        return sent


class Invoices:
    """Invoice texts and the sales of each day."""

    def __init__(self):
        self.texts = {}
        self.next_number = 1
        self.sales_by_day = {}

    def issue(self, order_id, customer, lines, prices):
        number = self.next_number
        self.next_number += 1
        rows = [f"Invoice {number} for order {order_id}, {customer}"]
        for sku, units, price in lines:
            rows.append(f"  {sku:<12}{units:>3} x {money(price):>10}")
        rows.append(f"  Subtotal {money(prices['subtotal']):>18}")
        if prices["discount"]:
            rows.append(f"  Discount {money(-prices['discount']):>18}")
        rows.append(f"  Tax      {money(prices['tax']):>18}")
        rows.append(f"  Delivery {money(prices['delivery']):>18}")
        rows.append(f"  Total    {money(prices['total']):>18}")
        self.texts[number] = "\n".join(rows)
        return number

    def text_of(self, number):
        return self.texts[number]

    def record(self, day, amount):
        self.sales_by_day[day] = self.sales_by_day.get(day, 0) + amount

    def report(self):
        return [f"day {day}: {money(self.sales_by_day[day])}" for day in sorted(self.sales_by_day)]

    def day_total(self, day):
        return self.sales_by_day.get(day, 0)


class OrderService:
    """A facade: the old public methods, each handing its work to the class that owns that job."""

    def __init__(self, pricing=None, stock=None, messages=None, invoices=None):
        self.pricing = pricing or Pricing()
        self.stock = stock or Stock()
        self.messages = messages or Messages()
        self.invoices = invoices or Invoices()
        self.orders = {}
        self.next_order_id = 1001

    # ---- catalogue and stock ------------------------------------------------------------------
    def add_product(self, sku, name, price):
        self.pricing.add_product(sku, name, price)

    def set_price(self, sku, price):
        self.pricing.set_price(sku, price)

    def remove_product(self, sku):
        self.pricing.remove_product(sku)
        self.stock.forget(sku)

    def price_of(self, sku):
        return self.pricing.price_of(sku)

    def add_coupon(self, code, kind, value, min_spend=0):
        self.pricing.add_coupon(code, kind, value, min_spend)

    def expire_coupon(self, code):
        self.pricing.expire_coupon(code)

    def coupon_is_valid(self, code, subtotal):
        return self.pricing.coupon_is_valid(code, subtotal)

    def quote(self, items, coupon=None, express=False):
        return self.pricing.quote(items, coupon, express)

    def restock(self, sku, units):
        if not self.pricing.has(sku):
            raise KeyError(sku)
        self.stock.restock(sku, units)

    def available(self, sku):
        return self.stock.available(sku)

    def units_held(self):
        return self.stock.units_held()

    def low_stock_report(self):
        return [f"{sku} {self.pricing.name_of(sku)}: {units} left" for sku, units in self.stock.low()]

    # ---- orders -------------------------------------------------------------------------------
    def place_order(self, customer, email, phone, items, coupon=None, express=False, day=1):
        if not items:
            raise ValueError("an order needs at least one item")
        if not re.fullmatch(r"\+?\d{10,13}", phone):
            raise ValueError(f"bad phone number {phone}")
        for sku, units in items:
            if not self.pricing.has(sku):
                raise ValueError(f"unknown product {sku}")
            if units <= 0:
                raise ValueError(f"bad quantity {units} for {sku}")
            self.stock.check(sku, units)
        order_id = self.next_order_id
        self.next_order_id += 1
        self.stock.hold(order_id, items)
        prices = self.pricing.quote(items, coupon, express)
        order = {"id": order_id, "customer": customer, "email": email, "phone": phone, "items": list(items),
                 "coupon": coupon, "express": express, "day": day, "status": "placed", "prices": prices,
                 "tracking": None}
        self.orders[order_id] = order
        self.invoices.record(day, prices["total"])
        self.messages.sms(phone, f"order {order_id} placed, {money(prices['total'])} to pay.")
        lines = [(sku, units, self.pricing.price_of(sku)) for sku, units in items]
        body = f"Hello {customer},\nwe have your order {order_id}.\n"
        for sku, units, price in lines:
            body += f"  {units} x {self.pricing.name_of(sku)}: {money(price * units)}\n"
        body += f"Total {money(prices['total'])}" + (" (express delivery)" if express else "") + "\n"
        self.messages.email(email, f"Your order {order_id}\n{body}")
        order["invoice"] = self.invoices.issue(order_id, customer, lines, prices)
        return order_id

    def cancel_order(self, order_id, reason, day=1):
        order = self.orders.get(order_id)
        if order is None:
            raise KeyError(order_id)
        if order["status"] == "shipped":
            raise ValueError(f"order {order_id} has shipped and cannot be cancelled")
        if order["status"] == "cancelled":
            return False
        self.stock.release(order_id)
        order["status"] = "cancelled"
        order["reason"] = reason
        total = order["prices"]["total"]
        self.invoices.record(day, -total)
        self.messages.sms(order["phone"], f"order {order_id} cancelled, {money(total)} refunded.")
        self.messages.email(order["email"], f"Order {order_id} cancelled\nReason: {reason}\nRefund {money(total)}\n")
        return True

    def ship_order(self, order_id, tracking):
        order = self.orders[order_id]
        if order["status"] != "placed":
            raise ValueError(f"order {order_id} is {order['status']}")
        order["status"] = "shipped"
        order["tracking"] = tracking
        self.stock.settle(order_id)
        self.messages.sms(order["phone"], f"order {order_id} shipped, tracking {tracking}.")
        self.messages.email(order["email"], f"Order {order_id} is on its way\nTracking: {tracking}\n")

    def status_of(self, order_id):
        return self.orders[order_id]["status"]

    def orders_of(self, customer):
        return [o["id"] for o in self.orders.values() if o["customer"] == customer]

    def customer_spend(self, customer):
        return sum(o["prices"]["total"] for o in self.orders.values()
                   if o["customer"] == customer and o["status"] != "cancelled")

    def refund_amount(self, order_id):
        order = self.orders[order_id]
        return order["prices"]["total"] if order["status"] == "cancelled" else 0

    # ---- messages, invoices and reports -------------------------------------------------------
    def set_sms(self, enabled):
        self.messages.set_sms(enabled)

    def messages_for(self, to):
        return self.messages.messages_for(to)

    def clear_outbox(self):
        return self.messages.clear()

    def resend_invoice(self, order_id):
        order = self.orders[order_id]
        self.messages.email(order["email"], self.invoices.text_of(order["invoice"]))

    def invoice_text(self, number):
        return self.invoices.text_of(number)

    def sales_report(self):
        return self.invoices.report()

    def day_total(self, day):
        return self.invoices.day_total(day)

    def best_sellers(self, n=3):
        units = {}
        for order in self.orders.values():
            if order["status"] == "cancelled":
                continue
            for sku, count in order["items"]:
                units[sku] = units.get(sku, 0) + count
        ranked = sorted(units.items(), key=lambda kv: (-kv[1], kv[0]))
        return [f"{self.pricing.name_of(sku)}: {count}" for sku, count in ranked[:n]]

    def export_report_json(self):
        report = {
            "orders": len(self.orders),
            "cancelled": sum(1 for o in self.orders.values() if o["status"] == "cancelled"),
            "shipped": sum(1 for o in self.orders.values() if o["status"] == "shipped"),
            "units_held": self.units_held(),
            "sales": {str(day): paise for day, paise in sorted(self.invoices.sales_by_day.items())},
            "low_stock": self.low_stock_report(),
        }
        return json.dumps(report, sort_keys=True)