মজবুত API ডিজাইন: পেজিনেশন, রেট লিমিট, রিট্রাই ও ভার্সনিং

mkdir -p ~/projects/robust-api
cd ~/projects/robust-api
uv init --bare
uv add -q "fastapi[standard]" requests
# uv-এর বদলে pip দিয়ে:
#   python -m venv .venv && source .venv/bin/activate
#   pip install -q "fastapi[standard]" requests

Initialized project `robust-api`

# file: items.py
import base64
import json
from typing import Annotated

from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel

router = APIRouter()

ITEMS = [{"id": number, "title": f"Note {number}"} for number in range(1, 26)]


class Item(BaseModel):
    id: int
    title: str


class NewItem(BaseModel):
    title: str


class OffsetPage(BaseModel):
    items: list[Item]
    total: int
    offset: int
    limit: int


class CursorPage(BaseModel):
    items: list[Item]
    next_cursor: str | None


def newest_first():
    return sorted(ITEMS, key=lambda item: item["id"], reverse=True)


def encode_cursor(last_id: int) -> str:
    return base64.urlsafe_b64encode(json.dumps({"last_id": last_id}).encode()).decode()


def decode_cursor(cursor: str) -> int:
    try:
        return int(json.loads(base64.urlsafe_b64decode(cursor.encode()))["last_id"])
    except (ValueError, KeyError, TypeError):
        raise HTTPException(400, "Invalid cursor")


@router.get("/items", response_model=OffsetPage)
def list_items(
    offset: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 10,
):
    rows = newest_first()
    return {"items": rows[offset:offset + limit], "total": len(rows), "offset": offset, "limit": limit}


@router.get("/feed", response_model=CursorPage)
def feed(cursor: str | None = None, limit: Annotated[int, Query(ge=1, le=100)] = 10):
    last_id = decode_cursor(cursor) if cursor else None
    rows = [row for row in newest_first() if last_id is None or row["id"] < last_id]
    page = rows[:limit]
    has_more = len(rows) > limit
    return {"items": page, "next_cursor": encode_cursor(page[-1]["id"]) if has_more else None}


@router.post("/items", status_code=201, response_model=Item)
def create_item(body: NewItem):
    item = {"id": max(row["id"] for row in ITEMS) + 1, "title": body.title}
    ITEMS.append(item)
    return item

# file: main.py
from fastapi import FastAPI

import items

app = FastAPI(title="Robust API demo", version="1.0.0")
app.include_router(items.router)


@app.get("/health")
def health():
    return {"status": "ok"}

uv run fastapi run main.py --port 8000 > server.log 2>&1 &
echo $! > server.pid
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health

{"status":"ok"}

import requests

APP = "http://localhost:8000"
PRACTICE = "http://localhost:8800"

first = requests.get(f"{APP}/items", params={"offset": 0, "limit": 5}).json()
print("offset page 1:", [item["id"] for item in first["items"]], "total", first["total"])

feed_first = requests.get(f"{APP}/feed", params={"limit": 5}).json()
print("cursor page 1:", [item["id"] for item in feed_first["items"]])
print("next_cursor:", feed_first["next_cursor"])

offset page 1: [25, 24, 23, 22, 21] total 25
cursor page 1: [25, 24, 23, 22, 21]
next_cursor: eyJsYXN0X2lkIjogMjF9

requests.post(f"{APP}/items", json={"title": "Note 26, added while you were reading"})

second_offset = requests.get(f"{APP}/items", params={"offset": 5, "limit": 5}).json()
print("offset page 2:", [item["id"] for item in second_offset["items"]])

second_cursor = requests.get(f"{APP}/feed", params={"limit": 5, "cursor": feed_first["next_cursor"]}).json()
print("cursor page 2:", [item["id"] for item in second_cursor["items"]])

offset page 2: [21, 20, 19, 18, 17]
cursor page 2: [20, 19, 18, 17, 16]

-- offset: ডেটাবেস আগে ১০০০০০টা রো পড়ে ফেলে দেয়
SELECT id, title FROM items ORDER BY id DESC LIMIT 10 OFFSET 100000;

-- cursor ("keyset"): ইনডেক্সের নির্দিষ্ট জায়গায় সরাসরি লাফ
SELECT id, title FROM items WHERE id < :last_id ORDER BY id DESC LIMIT 11;  -- ১১ = limit + 1

def iter_feed(base_url, limit=10):
    cursor = None
    while True:
        params = {"limit": limit}
        if cursor:
            params["cursor"] = cursor
        page = requests.get(f"{base_url}/feed", params=params, timeout=5).json()
        yield from page["items"]
        cursor = page["next_cursor"]
        if cursor is None:
            return

ids = [item["id"] for item in iter_feed(APP, limit=10)]
print(len(ids), ids[:3], ids[-1], len(set(ids)) == len(ids))

26 [26, 25, 24] 1 True

for number in range(1, 5):
    response = requests.get(f"{PRACTICE}/v1/limited", headers={"X-Client-Id": "robust-demo"})
    print(number, response.status_code, response.json(), "Retry-After:", response.headers.get("Retry-After"))

1 200 {'ok': True, 'remaining': 2} Retry-After: None
2 200 {'ok': True, 'remaining': 1} Retry-After: None
3 200 {'ok': True, 'remaining': 0} Retry-After: None
4 429 {'detail': 'Too many requests'} Retry-After: 10

import random
import time

def backoff_ceiling(attempt, base=0.5, cap=8.0):
    return min(cap, base * 2 ** attempt)

def backoff_delay(attempt, base=0.5, cap=8.0, rng=random):
    return rng.uniform(0, backoff_ceiling(attempt, base, cap))

print("ceilings:", [backoff_ceiling(a) for a in range(7)])
seeded = random.Random(7)
print("full jitter:", [round(backoff_delay(a, rng=seeded), 2) for a in range(7)])

ceilings: [0.5, 1.0, 2.0, 4.0, 8.0, 8.0, 8.0]
full jitter: [0.16, 0.15, 1.3, 0.29, 4.29, 2.93, 0.46]

RETRY_STATUS = {429, 500, 502, 503, 504}
IDEMPOTENT_METHODS = {"GET", "HEAD", "PUT", "DELETE", "OPTIONS"}


class RetryBudget:
    def __init__(self, ratio=0.1, reserve=2):
        self.ratio, self.cap, self.tokens = ratio, float(reserve), float(reserve)

    def deposit(self):
        self.tokens = min(self.cap, self.tokens + self.ratio)

    def withdraw(self):
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False


def call_with_retries(session, method, url, *, max_attempts=4, base=0.5, cap=8.0,
                      budget=None, sleep=time.sleep, rng=random, waits=None, **kwargs):
    kwargs.setdefault("timeout", (3, 10))
    safe = method.upper() in IDEMPOTENT_METHODS or "Idempotency-Key" in kwargs.get("headers", {})
    if not safe:
        max_attempts = 1
    if budget is not None:
        budget.deposit()
    for attempt in range(max_attempts):
        try:
            response, error = session.request(method, url, **kwargs), None
            if response.status_code not in RETRY_STATUS:
                return response
        except (requests.ConnectionError, requests.Timeout) as exc:
            response, error = None, exc
        if attempt == max_attempts - 1 or (budget is not None and not budget.withdraw()):
            if error is not None:
                raise error
            return response
        delay = backoff_delay(attempt, base, cap, rng)
        if response is not None and "Retry-After" in response.headers:
            delay = max(delay, float(response.headers["Retry-After"]))
        if waits is not None:
            waits.append(round(delay, 2))
        sleep(delay)

session = requests.Session()

waits = []
response = call_with_retries(
    session, "GET", f"{PRACTICE}/v1/flaky", params={"key": "robust-1", "fail_times": 2},
    sleep=lambda seconds: None, rng=random.Random(7), waits=waits,
)
print(response.status_code, response.json())
print("would have waited:", waits)

200 {'ok': True, 'attempt': 3}
would have waited: [0.16, 0.15]

waits = []
response = call_with_retries(
    session, "GET", f"{PRACTICE}/v1/limited", headers={"X-Client-Id": "robust-demo"},
    max_attempts=3, sleep=lambda seconds: None, rng=random.Random(7), waits=waits,
)
print(response.status_code, "after waits", waits)

429 after waits [10.0, 10.0]

budget = RetryBudget(ratio=0.1, reserve=2)
retries = []
for _ in range(20):
    call_with_retries(session, "GET", f"{PRACTICE}/v1/status/503", max_attempts=4,
                      budget=budget, sleep=lambda seconds: None, waits=retries)
print("calls: 20, retries actually made:", len(retries), "(without a budget: 60)")

calls: 20, retries actually made: 3 (without a budget: 60)

# file: payments.py
import hashlib
import json
import threading
from typing import Annotated

from fastapi import APIRouter, Header, HTTPException, Response
from pydantic import BaseModel, Field

router = APIRouter()

PAYMENTS: list[dict] = []
SEEN: dict[str, dict] = {}
LOCK = threading.Lock()


class PaymentIn(BaseModel):
    amount: int = Field(gt=0, description="Amount in poisha")
    currency: str = "BDT"
    note: str = ""


def fingerprint(payment: PaymentIn) -> str:
    return hashlib.sha256(json.dumps(payment.model_dump(), sort_keys=True).encode()).hexdigest()


@router.post("/payments", status_code=201)
def create_payment(
    payment: PaymentIn,
    response: Response,
    idempotency_key: Annotated[str | None, Header()] = None,
):
    if not idempotency_key:
        raise HTTPException(400, "The Idempotency-Key header is required")
    digest = fingerprint(payment)
    with LOCK:
        saved = SEEN.get(idempotency_key)
        if saved is not None:
            if saved["fingerprint"] != digest:
                raise HTTPException(422, "This Idempotency-Key was already used with a different request")
            response.headers["Idempotent-Replayed"] = "true"
            return saved["body"]
        body = {"id": f"pay_{len(PAYMENTS) + 1:04d}", **payment.model_dump()}
        PAYMENTS.append(body)
        SEEN[idempotency_key] = {"fingerprint": digest, "body": body}
        return body


@router.get("/payments")
def list_payments():
    return {"count": len(PAYMENTS), "payments": PAYMENTS}

POST /jobs                       client asks for the work
  <- 202 Accepted                 "I took it, it is not done"
     Location: /jobs/job_001      where to look
     Retry-After: 1               how soon to look

GET /jobs/job_001                client polls
  <- 200 {"status": "running", "progress": 0.5}
GET /jobs/job_001
  <- 200 {"status": "succeeded", "result": {...}}

# file: jobs.py
import asyncio
import itertools

from fastapi import APIRouter, HTTPException, Response
from pydantic import BaseModel, Field

router = APIRouter()

JOBS: dict[str, dict] = {}
TASKS: set[asyncio.Task] = set()
COUNTER = itertools.count(1)


class JobIn(BaseModel):
    documents: list[str] = Field(min_length=1, max_length=100)


async def run_job(job_id: str, documents: list[str]):
    job = JOBS[job_id]
    job["status"] = "running"
    for number in range(1, len(documents) + 1):
        await asyncio.sleep(0.05)
        job["progress"] = round(number / len(documents), 2)
    job["status"] = "succeeded"
    job["result"] = {"embedded": len(documents), "dimensions": 8}


@router.post("/jobs", status_code=202)
async def create_job(body: JobIn, response: Response):
    job_id = f"job_{next(COUNTER):03d}"
    JOBS[job_id] = {"id": job_id, "status": "queued", "progress": 0.0, "result": None}
    task = asyncio.create_task(run_job(job_id, body.documents))
    TASKS.add(task)
    task.add_done_callback(TASKS.discard)
    response.headers["Location"] = f"/jobs/{job_id}"
    response.headers["Retry-After"] = "1"
    return dict(JOBS[job_id])


@router.get("/jobs/{job_id}")
def get_job(job_id: str):
    job = JOBS.get(job_id)
    if job is None:
        raise HTTPException(404, f"Job {job_id} not found")
    return job

# file: versions.py
from fastapi import APIRouter, Depends, Response

DEPRECATION = "@1767225600"
SUNSET = "Thu, 31 Dec 2026 23:59:59 GMT"


def mark_deprecated(response: Response):
    response.headers["Deprecation"] = DEPRECATION
    response.headers["Sunset"] = SUNSET
    response.headers["Link"] = '</v2/profile>; rel="successor-version"'


v1 = APIRouter(prefix="/v1", dependencies=[Depends(mark_deprecated)])
v2 = APIRouter(prefix="/v2")


@v1.get("/profile")
def profile_v1():
    return {"id": 1, "name": "Nadia Rahman"}


@v2.get("/profile")
def profile_v2():
    return {"id": 1, "first_name": "Nadia", "last_name": "Rahman"}

# file: main.py
from fastapi import FastAPI

import items
import jobs
import payments
import versions

app = FastAPI(title="Robust API demo", version="1.1.0")
app.include_router(items.router)
app.include_router(payments.router)
app.include_router(jobs.router)
app.include_router(versions.v1)
app.include_router(versions.v2)


@app.get("/health")
def health():
    return {"status": "ok"}

kill $(cat server.pid)
while curl -s -o /dev/null http://localhost:8000/health; do sleep 0.1; done
uv run fastapi run main.py --port 8000 > server.log 2>&1 &
echo $! > server.pid
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health

{"status":"ok"}

curl -s -i http://localhost:8000/v1/profile | tr -d "\r" | grep -i -E "^(HTTP|deprecation|sunset|link|\{)"
echo
curl -s http://localhost:8000/v2/profile

HTTP/1.1 200 OK
deprecation: @1767225600
sunset: Thu, 31 Dec 2026 23:59:59 GMT
link: </v2/profile>; rel="successor-version"
{"id":1,"name":"Nadia Rahman"}

{"id":1,"first_name":"Nadia","last_name":"Rahman"}

import uuid

key = "order-1001"   # আসল কোডে: str(uuid.uuid4()), প্রতি অপারেশনে একবার বানানো
body = {"amount": 25000, "note": "Course fee"}

first = requests.post(f"{APP}/payments", json=body, headers={"Idempotency-Key": key})
retry = requests.post(f"{APP}/payments", json=body, headers={"Idempotency-Key": key})
print(first.status_code, first.json()["id"], first.headers.get("Idempotent-Replayed"))
print(retry.status_code, retry.json()["id"], retry.headers.get("Idempotent-Replayed"))

changed = requests.post(f"{APP}/payments", json={**body, "amount": 99999}, headers={"Idempotency-Key": key})
print(changed.status_code, changed.json()["detail"])

missing = requests.post(f"{APP}/payments", json=body)
print(missing.status_code, missing.json()["detail"])
print("payments stored:", requests.get(f"{APP}/payments").json()["count"])

201 pay_0001 None
201 pay_0001 true
422 This Idempotency-Key was already used with a different request
400 The Idempotency-Key header is required
payments stored: 1

response = call_with_retries(
    requests, "POST", f"{APP}/payments", json=body, headers={"Idempotency-Key": key},
    sleep=lambda seconds: None,
)
print(response.status_code, response.json()["id"], response.headers.get("Idempotent-Replayed"))

201 pay_0001 true

created = requests.post(f"{APP}/jobs", json={"documents": ["a", "b", "c", "d"]})
print(created.status_code, created.headers["Location"], created.headers["Retry-After"])
print(created.json())

def wait_for_job(base_url, location, interval=0.1, timeout=5.0):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        job = requests.get(base_url + location, timeout=5).json()
        if job["status"] in ("succeeded", "failed"):
            return job
        time.sleep(interval)
    raise TimeoutError(f"{location} did not finish in {timeout} seconds")

print(wait_for_job(APP, created.headers["Location"]))
print(requests.get(f"{APP}/jobs/job_999").status_code)

202 /jobs/job_001 1
{'id': 'job_001', 'status': 'queued', 'progress': 0.0, 'result': None}
{'id': 'job_001', 'status': 'succeeded', 'progress': 1.0, 'result': {'embedded': 4, 'dimensions': 8}}
404

spec = requests.get(f"{APP}/openapi.json").json()
print(spec["openapi"], spec["info"]["title"], spec["info"]["version"])
for path, methods in spec["paths"].items():
    print(f"{path:<18}", sorted(method.upper() for method in methods))
print(sorted(spec["components"]["schemas"]["Item"]["properties"]))

3.1.0 Robust API demo 1.1.0
/items             ['GET', 'POST']
/feed              ['GET']
/payments          ['GET', 'POST']
/jobs              ['POST']
/jobs/{job_id}     ['GET']
/v1/profile        ['GET']
/v2/profile        ['GET']
/health            ['GET']
['id', 'title']

import copy

def operations(spec):
    for path, methods in spec["paths"].items():
        for method, operation in methods.items():
            yield (method.upper(), path), operation

def required_params(operation):
    return {p["name"] for p in operation.get("parameters", []) if p.get("required")}

def resolve(spec, schema):
    if "$ref" in schema:
        return spec["components"]["schemas"][schema["$ref"].split("/")[-1]]
    return schema

def field_names(spec, schema, prefix=""):
    schema = resolve(spec, schema)
    if schema.get("type") == "array":
        return field_names(spec, schema["items"], prefix)
    names = set()
    for name, sub in schema.get("properties", {}).items():
        names.add(prefix + name)
        names |= field_names(spec, sub, prefix + name + ".")
    return names

def response_fields(spec, operation):
    for code in ("200", "201", "202"):
        if code in operation["responses"]:
            content = operation["responses"][code].get("content", {}).get("application/json", {})
            return field_names(spec, content.get("schema", {}))
    return set()

def breaking_changes(old, new):
    problems = []
    new_operations = dict(operations(new))
    for (method, path), old_operation in operations(old):
        label = f"{method} {path}"
        new_operation = new_operations.get((method, path))
        if new_operation is None:
            problems.append(f"{label}: endpoint removed")
            continue
        for name in sorted(required_params(new_operation) - required_params(old_operation)):
            problems.append(f"{label}: new required parameter {name!r}")
        for name in sorted(response_fields(old, old_operation) - response_fields(new, new_operation)):
            problems.append(f"{label}: response field {name!r} removed")
    return problems

print("no change:", breaking_changes(spec, spec))

no change: []

changed = copy.deepcopy(spec)
del changed["paths"]["/feed"]                                          # এন্ডপয়েন্ট সরানো হলো
changed["paths"]["/items"]["get"]["parameters"].append(                # নতুন আবশ্যক প্যারামিটার
    {"name": "tenant", "in": "query", "required": True, "schema": {"type": "string"}})
properties = changed["components"]["schemas"]["Item"]["properties"]
properties["name"] = properties.pop("title")                           # ফিল্ডের নাম বদলানো হলো
changed["paths"]["/stats"] = {"get": {"responses": {"200": {"description": "ok"}}}}   # নতুন এন্ডপয়েন্ট

for problem in breaking_changes(spec, changed):
    print(problem)

GET /items: new required parameter 'tenant'
GET /items: response field 'items.title' removed
POST /items: response field 'title' removed
GET /feed: endpoint removed

import json

class ResponseTooLarge(Exception):
    pass

def fetch_json(method, url, *, max_bytes=100_000, timeout=(3, 5), deadline=10.0, **kwargs):
    started = time.monotonic()
    with requests.request(method, url, stream=True, timeout=timeout, allow_redirects=False, **kwargs) as response:
        response.raise_for_status()
        if "application/json" not in response.headers.get("Content-Type", ""):
            raise ValueError(f"expected JSON, got {response.headers.get('Content-Type')!r}")
        chunks, size = [], 0
        for chunk in response.iter_content(chunk_size=1024):
            size += len(chunk)
            if size > max_bytes:
                raise ResponseTooLarge(f"more than {max_bytes} bytes")
            if time.monotonic() - started > deadline:
                raise TimeoutError(f"took longer than {deadline} seconds")
            chunks.append(chunk)
    return json.loads(b"".join(chunks))

print(fetch_json("GET", f"{PRACTICE}/v1/courses/8"))

try:
    fetch_json("POST", f"{PRACTICE}/v1/echo", json={"blob": "x" * 5000}, max_bytes=1000)
except ResponseTooLarge as exc:
    print("ResponseTooLarge:", exc)

try:
    fetch_json("GET", f"{PRACTICE}/v1/slow", params={"seconds": 3}, timeout=(1, 0.5))
except requests.exceptions.Timeout as exc:
    print("Timeout:", type(exc).__name__)

{'id': 8, 'title': 'RAG Systems', 'level': 'advanced', 'hours': 10, 'rating': 4.9}
ResponseTooLarge: more than 1000 bytes
Timeout: ReadTimeout

from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, ValidationError

class Course(BaseModel):
    model_config = ConfigDict(extra="ignore")
    id: int
    title: str = Field(max_length=200)
    level: Literal["beginner", "intermediate", "advanced"]
    hours: int = Field(ge=0, le=1000)
    rating: float = Field(ge=0, le=5)

page = fetch_json("GET", f"{PRACTICE}/v1/courses", params={"per_page": 20})
courses = [Course(**row) for row in page["items"]]
print(len(courses), "valid courses, first:", courses[0].title)

try:
    Course(**fetch_json("GET", f"{PRACTICE}/v1/weather", params={"city": "Dhaka"}))
except ValidationError as exc:
    print("rejected, bad fields:", [error["loc"][0] for error in exc.errors()])

10 valid courses, first: How AI Works
rejected, bad fields: ['id', 'title', 'level', 'hours', 'rating']

# 1. Easy
pages, total, cursor = 0, 0, None
while True:
    params = {"limit": 7, **({"cursor": cursor} if cursor else {})}
    page = requests.get(f"{APP}/feed", params=params).json()
    pages += 1
    total += len(page["items"])
    cursor = page["next_cursor"]
    if cursor is None:
        break
print(pages, total)


# 2. Medium
one = requests.post(f"{APP}/payments", json={"amount": 100}, headers={"Idempotency-Key": "ex-2"})
two = requests.post(f"{APP}/payments", json={"amount": 200}, headers={"Idempotency-Key": "ex-2"})
print(one.status_code, one.json()["id"])
print(two.status_code, two.json()["detail"])


# 3. Medium
job = requests.post(f"{APP}/jobs", json={"documents": ["x", "y"]}).json()
interval, polls = 0.02, 0
while True:
    job = requests.get(f"{APP}/jobs/{job['id']}").json()
    polls += 1
    if job["status"] in ("succeeded", "failed"):
        break
    time.sleep(interval)
    interval = min(interval * 2, 0.2)
print(job["status"], job["result"], polls >= 1)


# 4. Hard
def additive_changes(old, new):
    found = []
    old_operations = dict(operations(old))
    for key, operation in operations(new):
        if key not in old_operations:
            found.append(f"{key[0]} {key[1]}: new endpoint")
        else:
            for name in sorted(response_fields(new, operation) - response_fields(old, old_operations[key])):
                found.append(f"{key[0]} {key[1]}: new response field {name!r}")
    return found

for line in additive_changes(spec, changed):
    print(line)

4 25
201 pay_0002
422 This Idempotency-Key was already used with a different request
succeeded {'embedded': 2, 'dimensions': 8} True
GET /items: new response field 'items.name'
POST /items: new response field 'name'
GET /stats: new endpoint

kill $(cat server.pid)