REST API design

import sqlite3
from decimal import Decimal, ROUND_HALF_UP

import pandas as pd
from flask import Flask, jsonify, request

base = "https://academy.cloudtechanalytics.com/datasets/invoicing/"
customers = pd.read_csv(base + "customers.csv")
raw = pd.read_csv(base + "invoices_raw.csv", dtype=str, keep_default_na=False).drop_duplicates()
lines = pd.read_csv(base + "invoice_lines.csv", dtype={"unit_price": str})

def kobo(naira):
    return int((Decimal(naira.replace("₦", "").replace(",", "")) * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))

def round_kobo(amount):
    return int(amount.quantize(Decimal("1"), rounding=ROUND_HALF_UP))

def invoice_total(line_rows, discount_pct, vat_exempt):
    subtotal = sum(q * p for q, p in line_rows)
    after = subtotal - round_kobo(Decimal(subtotal) * discount_pct / 100)
    return after + (0 if vat_exempt else round_kobo(after * Decimal("0.075")))

bad = set(lines.loc[lines["quantity"] <= 0, "invoice_id"])
clean = raw[raw["customer_id"].isin(set(customers["customer_id"])) & raw["issue_date"].str.match(r"^\d{4}-\d{2}-\d{2}
quot;)
            & (raw["discount_pct"].astype(int) <= 20) & (raw["due_date"] >= raw["issue_date"]) & ~raw["invoice_id"].isin(bad)]

db = sqlite3.connect(":memory:", check_same_thread=False)
db.row_factory = sqlite3.Row
db.executescript("""
CREATE TABLE customers (customer_id TEXT PRIMARY KEY, business_name TEXT, city TEXT, vat_exempt INTEGER, payment_terms_days INTEGER);
CREATE TABLE invoices (invoice_id TEXT PRIMARY KEY, customer_id TEXT REFERENCES customers, issue_date TEXT, due_date TEXT, discount_pct INTEGER);
CREATE TABLE invoice_lines (invoice_id TEXT, line_no INTEGER, description TEXT, quantity INTEGER, unit_price_kobo INTEGER, PRIMARY KEY (invoice_id, line_no));
CREATE TABLE payments (bank_reference TEXT PRIMARY KEY, invoice_id TEXT REFERENCES invoices, amount_kobo INTEGER CHECK (amount_kobo > 0));
CREATE INDEX idx_invoices_customer_date ON invoices (customer_id, issue_date);
""")
with db:
    db.executemany("INSERT INTO customers VALUES (?, ?, ?, ?, ?)", customers.itertuples(index=False))
    db.executemany("INSERT INTO invoices VALUES (?, ?, ?, ?, ?)", clean[["invoice_id", "customer_id", "issue_date", "due_date", "discount_pct"]].itertuples(index=False))
    db.executemany("INSERT INTO invoice_lines VALUES (?, ?, ?, ?, ?)",
                   [(r.invoice_id, r.line_no, r.description, r.quantity, kobo(r.unit_price)) for r in lines[lines["invoice_id"].isin(clean["invoice_id"])].itertuples()])
    db.executemany("INSERT INTO payments VALUES (?, ?, ?)",
                   [(f"EXPORT-{r.invoice_id}", r.invoice_id, kobo(r.amount_paid)) for r in clean.itertuples() if kobo(r.amount_paid) > 0])

def invoice_summary(invoice_id):
    inv = db.execute("SELECT i.*, c.vat_exempt FROM invoices i JOIN customers c USING (customer_id) WHERE invoice_id = ?", (invoice_id,)).fetchone()
    if inv is None:
        return None
    rows = db.execute("SELECT line_no, description, quantity, unit_price_kobo FROM invoice_lines WHERE invoice_id = ? ORDER BY line_no", (invoice_id,)).fetchall()
    total = invoice_total([(r["quantity"], r["unit_price_kobo"]) for r in rows], inv["discount_pct"], inv["vat_exempt"])
    paid = db.execute("SELECT COALESCE(SUM(amount_kobo), 0) FROM payments WHERE invoice_id = ?", (invoice_id,)).fetchone()[0]
    return {"invoice_id": invoice_id, "customer_id": inv["customer_id"], "due_date": inv["due_date"],
            "lines": [dict(r) for r in rows], "total_kobo": total, "paid_kobo": paid, "balance_kobo": total - paid}

app = Flask(__name__)

@app.get("/customers/<customer_id>/invoices")
def customer_invoices(customer_id):
    if db.execute("SELECT 1 FROM customers WHERE customer_id = ?", (customer_id,)).fetchone() is None:
        return jsonify(error="No such customer"), 404
    limit = min(int(request.args.get("limit", 10)), 50)
    offset = int(request.args.get("offset", 0))
    count = db.execute("SELECT COUNT(*) FROM invoices WHERE customer_id = ?", (customer_id,)).fetchone()[0]
    rows = db.execute("SELECT invoice_id, issue_date, due_date FROM invoices WHERE customer_id = ? ORDER BY issue_date DESC LIMIT ? OFFSET ?",
                      (customer_id, limit, offset)).fetchall()
    return jsonify(total=count, limit=limit, offset=offset, invoices=[dict(r) for r in rows])

@app.get("/invoices/<invoice_id>")
def get_invoice(invoice_id):
    summary = invoice_summary(invoice_id)
    return (jsonify(summary), 200) if summary else (jsonify(error="No such invoice"), 404)

@app.post("/invoices/<invoice_id>/payments")
def add_payment(invoice_id):
    summary = invoice_summary(invoice_id)
    if summary is None:
        return jsonify(error="No such invoice"), 404
    body = request.get_json(silent=True) or {}
    amount, reference = body.get("amount_kobo"), body.get("bank_reference")
    if not isinstance(amount, int) or amount <= 0 or not reference:
        return jsonify(error="Send a positive whole-number amount_kobo and a bank_reference"), 400
    if amount > summary["balance_kobo"]:
        return jsonify(error=f"Payment exceeds the balance of {summary['balance_kobo']} kobo"), 409
    try:
        with db:
            db.execute("INSERT INTO payments VALUES (?, ?, ?)", (reference, invoice_id, amount))
    except sqlite3.IntegrityError:
        return jsonify(error="This bank_reference has already been recorded"), 409
    return jsonify(invoice_summary(invoice_id)), 201, {"Location": f"/invoices/{invoice_id}"}

client = app.test_client()
r = client.get("/customers/C0145/invoices?limit=3")
print(r.status_code, {k: r.get_json()[k] for k in ("total", "limit", "offset")}, [i["invoice_id"] for i in r.get_json()["invoices"]])

200 {'total': 4, 'limit': 3, 'offset': 0} ['INV-100357', 'INV-100760', 'INV-100832']

unpaid = db.execute("""SELECT i.invoice_id FROM invoices i LEFT JOIN payments p USING (invoice_id)
                       WHERE p.invoice_id IS NULL ORDER BY i.invoice_id LIMIT 1""").fetchone()[0]
before = client.get(f"/invoices/{unpaid}").get_json()
print("Before:", before["total_kobo"], "total,", before["balance_kobo"], "balance")

steps = [
    ("pay 100,000 kobo", {"amount_kobo": 100_000, "bank_reference": "BNK-9001"}),
    ("same reference again", {"amount_kobo": 100_000, "bank_reference": "BNK-9001"}),
    ("more than the balance", {"amount_kobo": before["total_kobo"], "bank_reference": "BNK-9002"}),
    ("no amount", {"bank_reference": "BNK-9003"}),
]
for name, body in steps:
    r = client.post(f"/invoices/{unpaid}/payments", json=body)
    detail = r.get_json().get("error") or f"balance now {r.get_json()['balance_kobo']}, Location {r.headers['Location']}"
    print(f"{name:22} {r.status_code}  {detail}")
print("Unknown invoice:", client.get("/invoices/INV-000000").status_code)

Before: 3784484 total, 3784484 balance
pay 100,000 kobo       201  balance now 3684484, Location /invoices/INV-100001
same reference again   409  This bank_reference has already been recorded
more than the balance  409  Payment exceeds the balance of 3684484 kobo
no amount              400  Send a positive whole-number amount_kobo and a bank_reference
Unknown invoice: 404