# केस स्टडी: Payments और Items API — API Design और Versioning

Source: https://www.geekswithgeeks.com/hi/api-design/wrap-case-study

> इस कोर्स के outputs बनाने वाले पूरे demo API को देखें।

## Demo में क्या है

Demo लगभग 60 पंक्तियों का Python standard library कोड है। `GET /v1/items` और `GET /v2/items/{id}` **resources, cursor pagination, ETags और 304** दिखाते हैं। `/v1/items/{id}` `Deprecation`, `Sunset` और `Link` headers वाला **deprecated** आकार है जबकि `/v2/items/{id}` **उत्तराधिकारी** है। `POST /v1/payments` **idempotency keys**, **Location के साथ 201**, **400 और 422 problem responses** दिखाता है। `/limited` `429` और `Retry-After` के साथ **rate limiting** दिखाता है। Errors `application/problem+json` उपयोग करते हैं। यह सिखाने वाला server है, production कोड नहीं: इसमें authentication, persistence या concurrency नियंत्रण नहीं है, और असली सेवा तीनों जोड़ेगी, साथ में सुरक्षा खंड के object-level authorisation नियम भी।

## भरोसेमंद, विकसित होने योग्य API

Resource डिज़ाइन, स्पष्ट errors, सुरक्षित retries, सुरक्षा और versioning योजना साथ काम करते हैं।

![चार हिस्से: डिज़ाइन, व्यवहार, सुरक्षा, विकास।](assets/figures/api-design/section-8-map.svg) — चित्र 8.1 — डिज़ाइन, व्यवहार, सुरक्षा और विकास।

## Demo server (पूरा कोड)

`server.py` के रूप में सहेजें। यह सिर्फ़ standard library उपयोग करता है। इसे `from server import start; srv = start()` से शुरू करें और किसी भी HTTP client से बुलाएँ। इस कोर्स के सारे outputs इसी कोड को चलाने से आए।

```python
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

```

## Client helper

Demo snippets द्वारा उपयोग होने वाला `call()` helper: यह status, headers और parse की गई JSON body लौटाता है।

```python
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)

```

**Quiz:** Demo server जानबूझकर कौन-सी production ज़रूरतें छोड़ता है?

- [x] Authentication, persistence और concurrency नियंत्रण
- [ ] HTTP status codes
- [ ] JSON responses
- [ ] सभी headers

*Answer:* Authentication, persistence और concurrency नियंत्रण. यह API व्यवहार सिखाता है, इसलिए असली deployments को सुरक्षा और storage जोड़ने होंगे।