केस स्टडी: Payments और Items 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 योजना साथ काम करते हैं।

Demo server (पूरा कोड)

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

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 लौटाता है।

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)

त्वरित जाँच: Demo server जानबूझकर कौन-सी production ज़रूरतें छोड़ता है?

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