ওয়েব API ও REST: ডকুমেন্টেশন পড়া আর curl দিয়ে পরীক্ষা

mkdir -p ~/projects/practice-api
cd ~/projects/practice-api
python -m venv .venv
source .venv/bin/activate
pip install -q "fastapi[standard]"

"""BitByte Practice API: a small, offline web API for the Git, APIs and Backend tutorials.

Run it in its own terminal and leave it running:

    pip install "fastapi[standard]"
    fastapi run practice_api.py --port 8800

Then open http://localhost:8800/docs. Everything it returns is made up and the same
on every machine, so your output matches the tutorial's.
"""
import asyncio
import json
import time

from fastapi import FastAPI, Header, HTTPException, Query, Request, Response
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel, Field

app = FastAPI(title="BitByte Practice API", version="1.0.0")

API_KEY = "practice-key-123"
USERS = {"nadia": "s3cret-pass"}
TOKENS = {"token-nadia-001": "nadia"}

COURSES = [
    {"id": 1, "title": "How AI Works", "level": "beginner", "hours": 3, "rating": 4.7},
    {"id": 2, "title": "Python for AI", "level": "beginner", "hours": 12, "rating": 4.8},
    {"id": 3, "title": "Maths for AI", "level": "beginner", "hours": 8, "rating": 4.5},
    {"id": 4, "title": "SQL and Data Handling", "level": "beginner", "hours": 10, "rating": 4.6},
    {"id": 5, "title": "Git, APIs and Backend", "level": "beginner", "hours": 9, "rating": 4.6},
    {"id": 6, "title": "Prompt Engineering", "level": "intermediate", "hours": 5, "rating": 4.4},
    {"id": 7, "title": "LLM APIs", "level": "intermediate", "hours": 6, "rating": 4.7},
    {"id": 8, "title": "RAG Systems", "level": "advanced", "hours": 10, "rating": 4.9},
    {"id": 9, "title": "AI Agents", "level": "advanced", "hours": 12, "rating": 4.8},
    {"id": 10, "title": "LLMOps", "level": "advanced", "hours": 8, "rating": 4.3},
]

WEATHER = {
    "dhaka": {"city": "Dhaka", "temp_c": 31.5, "humidity": 78, "condition": "Humid"},
    "chattogram": {"city": "Chattogram", "temp_c": 29.8, "humidity": 82, "condition": "Rain"},
    "sylhet": {"city": "Sylhet", "temp_c": 27.4, "humidity": 85, "condition": "Rain"},
    "khulna": {"city": "Khulna", "temp_c": 32.1, "humidity": 70, "condition": "Sunny"},
    "rajshahi": {"city": "Rajshahi", "temp_c": 33.6, "humidity": 58, "condition": "Sunny"},
}

REVIEWS: list[dict] = []
RATE_WINDOW_SECONDS = 10
RATE_LIMIT = 3
_rate_hits: dict[str, list[float]] = {}
_flaky_calls: dict[str, int] = {}


class ReviewIn(BaseModel):
    course_id: int = Field(ge=1)
    rating: int = Field(ge=1, le=5)
    comment: str = Field(min_length=3, max_length=500)


class ChatMessage(BaseModel):
    role: str
    content: str


class ChatRequest(BaseModel):
    model: str = "practice-llm-1"
    messages: list[ChatMessage] = Field(min_length=1)
    stream: bool = False


def require_api_key(x_api_key: str | None) -> None:
    if x_api_key is None:
        raise HTTPException(401, "Missing API key: send it in the X-API-Key header")
    if x_api_key != API_KEY:
        raise HTTPException(403, "Invalid API key")


def require_bearer(authorization: str | None) -> str:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401, "Missing bearer token", headers={"WWW-Authenticate": "Bearer"})
    user = TOKENS.get(authorization.removeprefix("Bearer "))
    if user is None:
        raise HTTPException(401, "Invalid or expired token", headers={"WWW-Authenticate": "Bearer"})
    return user


@app.get("/")
def home():
    return {"name": "BitByte Practice API", "version": "1.0.0", "docs": "/docs"}


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


@app.get("/v1/courses")
def list_courses(
    level: str | None = None,
    page: int = Query(1, ge=1),
    per_page: int = Query(5, ge=1, le=20),
):
    items = [c for c in COURSES if level is None or c["level"] == level]
    start = (page - 1) * per_page
    chunk = items[start:start + per_page]
    return {
        "page": page,
        "per_page": per_page,
        "total": len(items),
        "next_page": page + 1 if start + per_page < len(items) else None,
        "items": chunk,
    }


@app.get("/v1/courses/{course_id}")
def get_course(course_id: int):
    for course in COURSES:
        if course["id"] == course_id:
            return course
    raise HTTPException(404, f"Course {course_id} not found")


@app.get("/v1/weather")
def weather(city: str):
    data = WEATHER.get(city.lower())
    if data is None:
        raise HTTPException(404, f"No weather data for {city!r}")
    return data


@app.post("/v1/reviews", status_code=201)
def create_review(review: ReviewIn, x_api_key: str | None = Header(None)):
    require_api_key(x_api_key)
    if not any(c["id"] == review.course_id for c in COURSES):
        raise HTTPException(404, f"Course {review.course_id} not found")
    saved = {"id": len(REVIEWS) + 1, **review.model_dump()}
    REVIEWS.append(saved)
    return saved


@app.post("/v1/token")
def token(form: dict):
    username, password = form.get("username"), form.get("password")
    if USERS.get(username) != password:
        raise HTTPException(401, "Wrong username or password")
    return {"access_token": "token-nadia-001", "token_type": "bearer", "expires_in": 3600}


@app.get("/v1/me")
def me(authorization: str | None = Header(None)):
    user = require_bearer(authorization)
    return {"username": user, "plan": "student"}


@app.get("/v1/limited")
def limited(x_client_id: str = Header("anonymous")):
    """Allows 3 requests per 10 seconds per X-Client-Id, then answers 429."""
    now = time.monotonic()
    hits = [t for t in _rate_hits.get(x_client_id, []) if now - t < RATE_WINDOW_SECONDS]
    if len(hits) >= RATE_LIMIT:
        retry_after = max(1, int(RATE_WINDOW_SECONDS - (now - hits[0])) + 1)
        _rate_hits[x_client_id] = hits
        return JSONResponse(
            {"detail": "Too many requests"}, status_code=429, headers={"Retry-After": str(retry_after)}
        )
    hits.append(now)
    _rate_hits[x_client_id] = hits
    return {"ok": True, "remaining": RATE_LIMIT - len(hits)}


@app.get("/v1/flaky")
def flaky(key: str, fail_times: int = Query(2, ge=0, le=10)):
    """Fails with 503 for the first `fail_times` calls with this key, then succeeds."""
    calls = _flaky_calls.get(key, 0) + 1
    _flaky_calls[key] = calls
    if calls <= fail_times:
        return JSONResponse({"detail": "Service temporarily unavailable"}, status_code=503)
    return {"ok": True, "attempt": calls}


@app.get("/v1/slow")
async def slow(seconds: float = Query(2.0, ge=0, le=30)):
    await asyncio.sleep(seconds)
    return {"waited_seconds": seconds}


@app.api_route("/v1/echo", methods=["GET", "POST", "PUT", "PATCH", "DELETE"])
async def echo(request: Request):
    body = await request.body()
    try:
        parsed = json.loads(body) if body else None
    except json.JSONDecodeError:
        parsed = body.decode(errors="replace")
    shown = {k: v for k, v in request.headers.items() if k in ("content-type", "authorization", "x-api-key", "user-agent", "accept")}
    return {"method": request.method, "query": dict(request.query_params), "headers": shown, "json": parsed}


@app.get("/v1/status/{code}")
def status(code: int):
    if not 200 <= code <= 599:
        raise HTTPException(400, "Status code must be between 200 and 599")
    return Response(status_code=code) if code in (204, 304) else JSONResponse({"status": code}, status_code=code)


def _fake_answer(messages: list[ChatMessage]) -> str:
    question = messages[-1].content.strip().rstrip("?")
    return f"This is a practice answer about: {question}. A real model would say more."


@app.post("/v1/chat")
async def chat(request: ChatRequest, authorization: str | None = Header(None)):
    """A fake LLM. Same shape as many real chat APIs; the answer is always predictable."""
    require_bearer(authorization)
    answer = _fake_answer(request.messages)
    prompt_tokens = sum(len(m.content.split()) for m in request.messages)
    completion_tokens = len(answer.split())
    usage = {"prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens,
             "total_tokens": prompt_tokens + completion_tokens}
    if not request.stream:
        return {"model": request.model, "message": {"role": "assistant", "content": answer}, "usage": usage}

    async def events():
        for word in answer.split(" "):
            yield f"data: {json.dumps({'delta': word + ' '})}\n\n"
            await asyncio.sleep(0.05)
        yield f"data: {json.dumps({'done': True, 'usage': usage})}\n\n"

    return StreamingResponse(events(), media_type="text/event-stream")

# দ্বিতীয় টার্মিনাল: এটা চালু রেখে দিন
cd ~/projects/practice-api
source .venv/bin/activate
fastapi run practice_api.py --port 8800

curl -s http://localhost:8800/health

{"status":"ok"}

curl -s http://localhost:8800/openapi.json | python -c '
import json, sys
spec = json.load(sys.stdin)
print("OpenAPI", spec["openapi"], "|", spec["info"]["title"], spec["info"]["version"])
for path, operations in spec["paths"].items():
    print(f"{path:26}", ", ".join(sorted(m.upper() for m in operations)))
'

OpenAPI 3.1.0 | BitByte Practice API 1.0.0
/                          GET
/health                    GET
/v1/courses                GET
/v1/courses/{course_id}    GET
/v1/weather                GET
/v1/reviews                POST
/v1/token                  POST
/v1/me                     GET
/v1/limited                GET
/v1/flaky                  GET
/v1/slow                   GET
/v1/echo                   DELETE, GET, PATCH, POST, PUT
/v1/status/{code}          GET
/v1/chat                   POST

curl -s "http://localhost:8800/v1/courses?per_page=3" | python -m json.tool

{
    "page": 1,
    "per_page": 3,
    "total": 10,
    "next_page": 2,
    "items": [
        {
            "id": 1,
            "title": "How AI Works",
            "level": "beginner",
            "hours": 3,
            "rating": 4.7
        },
        {
            "id": 2,
            "title": "Python for AI",
            "level": "beginner",
            "hours": 12,
            "rating": 4.8
        },
        {
            "id": 3,
            "title": "Maths for AI",
            "level": "beginner",
            "hours": 8,
            "rating": 4.5
        }
    ]
}

curl -s -w "  -> %{http_code}\n" http://localhost:8800/v1/reviews \
  --json '{"course_id": 2, "rating": 5, "comment": "Clear and practical"}'
curl -s -w "  -> %{http_code}\n" http://localhost:8800/v1/reviews -H "X-API-Key: wrong-key" \
  --json '{"course_id": 2, "rating": 5, "comment": "Clear and practical"}'
curl -s -w "  -> %{http_code}\n" http://localhost:8800/v1/reviews -H "X-API-Key: practice-key-123" \
  --json '{"course_id": 2, "rating": 9, "comment": "ok"}'

{"detail":"Missing API key: send it in the X-API-Key header"}  -> 401
{"detail":"Invalid API key"}  -> 403
{"detail":[{"type":"less_than_equal","loc":["body","rating"],"msg":"Input should be less than or equal to 5","input":9,"ctx":{"le":5}},{"type":"string_too_short","loc":["body","comment"],"msg":"String should have at least 3 characters","input":"ok","ctx":{"min_length":3}}]}  -> 422

export PRACTICE_API_KEY=practice-key-123
curl -s -i http://localhost:8800/v1/reviews -H "X-API-Key: $PRACTICE_API_KEY" \
  --json '{"course_id": 2, "rating": 5, "comment": "Clear and practical"}' | grep -v '^date' | tr -d '\r'

HTTP/1.1 201 Created
server: uvicorn
content-length: 65
content-type: application/json

{"id":1,"course_id":2,"rating":5,"comment":"Clear and practical"}

curl -s http://localhost:8800/v1/token --json '{"username": "nadia", "password": "s3cret-pass"}'

{"access_token":"token-nadia-001","token_type":"bearer","expires_in":3600}

curl -si http://localhost:8800/v1/me | grep -v '^date' | tr -d '\r'

HTTP/1.1 401 Unauthorized
server: uvicorn
www-authenticate: Bearer
content-length: 33
content-type: application/json

{"detail":"Missing bearer token"}

TOKEN=$(curl -s http://localhost:8800/v1/token \
  --json '{"username": "nadia", "password": "s3cret-pass"}' \
  | python -c 'import json, sys; print(json.load(sys.stdin)["access_token"])')
curl -s http://localhost:8800/v1/me -H "Authorization: Bearer $TOKEN"

{"username":"nadia","plan":"student"}

curl -s http://localhost:8800/v1/courses/99 > /dev/null
echo "without --fail-with-body: exit code $?"
curl -s --fail-with-body http://localhost:8800/v1/courses/99
echo " with --fail-with-body: exit code $?"

without --fail-with-body: exit code 0
{"detail":"Course 99 not found"} with --fail-with-body: exit code 22

curl -s -m 1 "http://localhost:8800/v1/slow?seconds=3"
echo "exit code $?"

exit code 28

curl -s -o dhaka.json -w "%{http_code} %{content_type}\n" "http://localhost:8800/v1/weather?city=Dhaka"
python -m json.tool dhaka.json

200 application/json
{
    "city": "Dhaka",
    "temp_c": 31.5,
    "humidity": 78,
    "condition": "Humid"
}

for i in 1 2 3 4; do
  curl -s -w "  -> %{http_code}\n" -H "X-Client-Id: nadia-laptop" http://localhost:8800/v1/limited
done
curl -si -H "X-Client-Id: nadia-laptop" http://localhost:8800/v1/limited | grep -i -E '^HTTP|^retry-after' | tr -d '\r'

{"ok":true,"remaining":2}  -> 200
{"ok":true,"remaining":1}  -> 200
{"ok":true,"remaining":0}  -> 200
{"detail":"Too many requests"}  -> 429
HTTP/1.1 429 Too Many Requests
retry-after: 10

curl -s -H "X-Client-Id: rafi-laptop" http://localhost:8800/v1/limited

{"ok":true,"remaining":2}

# 1
curl -s http://localhost:8800/openapi.json | python -c '
import json, sys
spec = json.load(sys.stdin)
for p in spec["paths"]["/v1/courses"]["get"]["parameters"]:
    print(p["name"], p["in"], "required" if p.get("required") else "optional")
'

level query optional
page query optional
per_page query optional

# 2
curl -s -w "  -> %{http_code}\n" http://localhost:8800/v1/reviews -H "X-API-Key: $PRACTICE_API_KEY" \
  --json '{"course_id": 99, "rating": 4, "comment": "Looking forward to it"}'

{"detail":"Course 99 not found"}  -> 404

# 3
for i in 1 2 3 4 5; do
  headers=$(curl -s -D - -o /dev/null -H "X-Client-Id: batch-job" http://localhost:8800/v1/limited | tr -d '\r')
  code=$(echo "$headers" | head -n 1 | cut -d ' ' -f 2)
  if [ "$code" = "429" ]; then
    wait=$(echo "$headers" | grep -i '^retry-after:' | cut -d ' ' -f 2)
    echo "call $i: $code, retry after $wait s"
  else
    echo "call $i: $code"
  fi
done

call 1: 200
call 2: 200
call 3: 200
call 4: 429, retry after 10 s
call 5: 429, retry after 10 s