Case Study: A Payments and Items API

What the demo contains

The demo is about 60 lines of Python standard library. GET /v1/items and GET /v2/items/{id} show resources, cursor pagination, ETags and 304. /v1/items/{id} is the deprecated shape with Deprecation, Sunset and Link headers while /v2/items/{id} is the successor. POST /v1/payments demonstrates idempotency keys, 201 with Location, 400 and 422 problem responses. /limited shows rate limiting with 429 and Retry-After. Errors use application/problem+json. It is a teaching server, not production code: it has no authentication, persistence or concurrency control, and a real service would add all three, plus the object-level authorisation rules from the security section.

A dependable, evolvable API

Resource design, clear errors, safe retries, security and a versioning plan work together.

The demo server (full code)

Save as server.py. It uses only the standard library. Start it with from server import start; srv = start() and call it with any HTTP client. All outputs in this course came from running this code.

import base64, hashlib, json, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs

ITEMS = [{"id": i, "first_name": n.split()[0], "last_name": n.split()[1], "plan": p}
         for i, (n, p) in enumerate([("Asha Rao", "pro"), ("Ravi Iyer", "free"), ("Meera Shah", "pro"),
                                      ("Kiran Das", "free"), ("Neha Jain", "pro")], start=1)]
PAYMENTS, HITS = {}, {"n": 0}

def encode_cursor(last_id): return base64.urlsafe_b64encode(json.dumps({"after": last_id}).encode()).decode()
def decode_cursor(c): return json.loads(base64.urlsafe_b64decode(c.encode()))["after"]
def etag_for(obj): return '"' + hashlib.sha256(json.dumps(obj, sort_keys=True).encode()).hexdigest()[:12] + '"'

class H(BaseHTTPRequestHandler):
    def log_message(self, *a): pass
    def send(self, status, body=None, headers=None, ctype="application/json"):
        raw = b"" if body is None else json.dumps(body).encode()
        self.send_response(status); self.send_header("Content-Type", ctype)
        for k, v in (headers or {}).items(): self.send_header(k, v)
        self.send_header("Content-Length", str(len(raw))); self.end_headers(); self.wfile.write(raw)
    def problem(self, status, title, detail, **extra):
        self.send(status, {"type": "about:blank", "title": title, "status": status, "detail": detail, **extra}, ctype="application/problem+json")
    def do_GET(self):
        u = urlparse(self.path); q = parse_qs(u.query); parts = u.path.strip("/").split("/")
        if parts[:2] == ["v1", "items"] and len(parts) == 2:
            limit = int(q.get("limit", ["2"])[0]); after = decode_cursor(q["cursor"][0]) if "cursor" in q else 0
            page = [i for i in ITEMS if i["id"] > after][:limit]
            nxt = encode_cursor(page[-1]["id"]) if len(page) == limit and page[-1]["id"] < ITEMS[-1]["id"] else None
            return self.send(200, {"data": [{"id": i["id"], "plan": i["plan"]} for i in page], "next_cursor": nxt})
        if parts[0] in ("v1", "v2") and parts[1] == "items" and len(parts) == 3:
            item = next((i for i in ITEMS if str(i["id"]) == parts[2]), None)
            if not item: return self.problem(404, "Not Found", f"Item {parts[2]} does not exist.")
            if parts[0] == "v1":
                body = {"id": item["id"], "name": item["first_name"] + " " + item["last_name"], "plan": item["plan"]}
                h = {"Deprecation": "true", "Sunset": "Wed, 31 Dec 2026 23:59:59 GMT", "Link": '</v2/items/%s>; rel="successor-version"' % item["id"]}
            else:
                body = {"id": item["id"], "first_name": item["first_name"], "last_name": item["last_name"], "plan": item["plan"]}; h = {}
            tag = etag_for(body); h["ETag"] = tag
            if self.headers.get("If-None-Match") == tag: return self.send(304, None, h)
            return self.send(200, body, h)
        if parts == ["limited"]:
            HITS["n"] += 1
            h = {"X-RateLimit-Limit": "3", "X-RateLimit-Remaining": str(max(0, 3 - HITS["n"]))}
            if HITS["n"] > 3: return self.send(429, {"title": "Too Many Requests"}, {**h, "Retry-After": "30"}, "application/problem+json")
            return self.send(200, {"ok": True}, h)
        return self.problem(404, "Not Found", "No such route.")
    def do_POST(self):
        n = int(self.headers.get("Content-Length", 0)); body = json.loads(self.rfile.read(n) or b"{}")
        if self.path == "/v1/payments":
            key = self.headers.get("Idempotency-Key")
            if not key: return self.problem(400, "Missing header", "Idempotency-Key is required for POST /v1/payments.")
            if body.get("amount", 0) <= 0: return self.problem(422, "Validation failed", "amount must be positive.", errors=[{"field": "amount", "message": "must be > 0"}])
            if key in PAYMENTS: return self.send(200, PAYMENTS[key], {"Idempotent-Replayed": "true"})
            PAYMENTS[key] = {"id": "pay_%d" % (len(PAYMENTS) + 1), "amount": body["amount"], "status": "succeeded"}
            return self.send(201, PAYMENTS[key], {"Location": "/v1/payments/" + PAYMENTS[key]["id"]})
        return self.problem(404, "Not Found", "No such route.")

def start():
    srv = ThreadingHTTPServer(("127.0.0.1", 0), H); threading.Thread(target=srv.serve_forever, daemon=True).start(); return srv

The client helper

The call() helper used by the demo snippets: it returns the status, headers and parsed JSON body.

import json, urllib.request, urllib.error
from server import start
srv = start(); base = f"http://127.0.0.1:{srv.server_address[1]}"
def call(method, path, body=None, headers=None):
    req = urllib.request.Request(base + path, method=method, headers=headers or {}, data=None if body is None else json.dumps(body).encode())
    try:
        r = urllib.request.urlopen(req)
    except urllib.error.HTTPError as e:
        r = e
    raw = r.read(); return r.status, dict(r.headers), (json.loads(raw) if raw else None)

Quick check: Which production concerns does the demo server deliberately omit?

• Authentication, persistence and concurrency control
• HTTP status codes
• JSON responses
• All headers