Lesson 3 — Beginner

Building APIs with FastAPI

Ab tum server ki taraf khade ho. FastAPI tumse sirf function ka signature maangta hai — baaki validation, serialization aur documentation khud nikaal leta. Is lesson mein routes aur parameters, Pydantic models, dependency injection, sahi status codes aur clean errors, async mein external calls, startup/config/middleware, aur TestClient se endpoint testing dekhoge.

FastAPI kyun — aur kaise kaam karta hai

Core idea: function ka signature hi API contract hai. Path parameter, query parameter, request body, return type — sab wahi se dikhte hain. Ek jagah likho, teen kaam (validate, serialize, document) automatic ho jaate hain.

Routes, path/query parameters aur bodies

from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

class OrderIn(BaseModel):
    item: str
    qty: int = 1

@app.get("/orders/{order_id}")            # path parameter
def get_order(order_id: int, verbose: bool = False):   # query parameter
    return {"id": order_id, "verbose": verbose}

@app.get("/orders")                       # list + filters
def list_orders(status: str = Query("pending"), page: int = 1):
    return {"status": status, "page": page}

@app.post("/orders", status_code=201)     # body + custom status
def create_order(order: OrderIn):
    return {"id": 981, **order.model_dump()}

• Path params /{order_id}: resource ka identity. Type hint laga diya toh FastAPI khud coerce+validate karta hai — URL mein "abc" daala toh 422, tumhara code crash nahi.
• Query params: filters, pagination, options. Default value = automatically optional. Query() se description/limit bhi lag sakte hain.
• Body: Pydantic model ka parameter likho — FastAPI JSON padhega, validate karega aur typed object dega.
• Decorator hi routing hai: @app.get/post/put/patch/delete. Return dict ho toh JSON ban jaata hai, response_model se shape control hoti hai.

Pydantic — validation jo khud likhne ki zarurat nahi

from pydantic import BaseModel, Field

class UserIn(BaseModel):
    name: str = Field(min_length=1, max_length=80)
    email: str
    age: int = Field(default=18, ge=0, le=120)

# Aise payload par FastAPI khud 422 + structured error body bhejta hai:
# {"name": "", "email": "not-an-email", "age": 500}

# Output model alag rakho — andar ki har field bahar nahi jaati
class UserOut(BaseModel):
    name: str
    email: str

Validation ka rule: input model (UserIn) aur output model (UserOut) alag rakho. Andar ke har field ko bahar bhejna zaroori nahi — password hash, internal flags, DB ids kabhi response mein nahi jaane chahiye. Response model wahi chhaanta hai jo dikhana hai.

Dependency injection — reusable pieces

from fastapi import Depends, FastAPI, Header, HTTPException

# app = FastAPI() pehle se bana hai

def current_user(authorization: str = Header(default="")):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing token")
    return parse_token(authorization)      # tumhara token parsing logic

def get_db():
    conn = connect_database()              # tumhara connection logic
    try:
        yield conn                          # use ke liye do
    finally:
        conn.close()                        # kaam khatam, band karo

@app.get("/me")
def me(user = Depends(current_user), db = Depends(get_db)):
    return {"user": user}

• Depends ka matlab: ye kaam pehle kar do aur result do. Auth, DB connection, config, pagination — har reusable cheez ek dependency.
• yield wali dependency setup + cleanup guarantee karti hai (finally close) — resource leak nahi hota.
• Ek jagah logic, har jagah use: auth 5 routes mein chahiye toh 5 baar copy nahi — ek dependency, 5 Depends.
• Testing mein swap: dependency_overrides se asli DB ki jagah in-memory fake laga sakte ho — tests production se jud nahi jaate.

Status codes aur clean exception handling

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

# app = FastAPI() pehle se hai

@app.get("/orders/{order_id}")
def get_order(order_id: int):
    order = find_order(order_id)
    if order is None:
        raise HTTPException(status_code=404, detail="Order not found")
    return order

# Unexpected errors ko bhi clean JSON mein badlo
@app.exception_handler(ValueError)
def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(status_code=400, content={"error": str(exc), "code": "bad_input"})

# Consistent shape har error mein: {"error": "...", "code": "..."}

• HTTPException se hi status bhejo: error ke liye 200 + {"error": ...} sabse ganda pattern hai — client ko status se pata chalta hai, body se detail.
• Har error ka ek hi shape rakho ({error, code}) — client parsing logic ek baar likhega, docs bhi saaf rahenge.
• Exception handlers se unexpected errors (ValueError, DB down) ko bhi controlled response mein badal sakte ho.
• Stack trace client ko kabhi mat dikhao: traceback server log mein, client ko sirf detail + request id.

Async endpoints aur external services

import httpx
from fastapi import Depends, FastAPI, HTTPException

# app = FastAPI(lifespan=lifespan) — client lifespan mein bana hai (next section)

async def get_client():
    return app.state.http       # lifespan mein bana hua AsyncClient

@app.get("/weather/{city}")
async def weather(city: str, client: httpx.AsyncClient = Depends(get_client)):
    try:
        r = await client.get(
            "https://api.example.com/weather",
            params={"city": city},
            timeout=10.0,
        )
    except httpx.TimeoutException:
        raise HTTPException(status_code=504, detail="Upstream timeout")
    r.raise_for_status()
    return r.json()

• async def endpoint mein sirf await wale calls rakho. Andar requests.get() daala toh event loop ruk jaayega — ek slow upstream tumhare saare users ko block karega.
• Async endpoint + async client = ek thread mein bahut saare upstream calls overlap. LLM API, weather API, third-party calls — sab isi pattern se.
• Sync endpoints (def) bhi valid hain — FastAPI unhe threadpool mein chalata hai. CPU thoda sa kaam ya sync-only library theek hai; lamba blocking work alag worker/queue mein daalo.
• Timeout aur error dono handle karo — upstream ka hang tumhara hang nahi banna chahiye. Timeout ko 504 mein badalna ek achha convention hai.

Startup, config aur middleware

import os
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(base_url=os.environ["UPSTREAM_URL"])
    print("startup: resources ready")
    yield
    await app.state.http.aclose()
    print("shutdown: resources closed")

app = FastAPI(lifespan=lifespan)

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Request-Id"] = request.headers.get("X-Request-Id", "generated-123")
    return response

• Lifespan = startup/shutdown. Connections, pools, clients yahan banao aur yahan band karo — har request pe naya banana sabse common performance bug hai.
• Config env vars se: os.environ ya Pydantic Settings. DATABASE_URL/API_KEY hardcode kabhi nahi — local aur production alag values lete hain.
• Middleware har request se guzarta hai — request id, timing, auth header, logging. Heavy kaam middleware mein mat rakho, warna har request mehngi ho jaayegi.
• Startup pe fail hona better hai — env var missing ho toh chup-chaap 500 dene ke bajaye startup par clear error aa jaaye.

Endpoints ka testing — TestClient

from fastapi.testclient import TestClient

client = TestClient(app)   # with TestClient(app) likho toh lifespan bhi chalta hai

def test_create_order():
    r = client.post("/orders", json={"item": "gpu-hours", "qty": 2})
    assert r.status_code == 201
    assert r.json()["qty"] == 2

def test_missing_field_is_422():
    r = client.post("/orders", json={"item": "gpu-hours"})
    assert r.status_code == 422

def test_missing_token_is_401():
    r = client.get("/me")
    assert r.status_code == 401

# chalao: uv run pytest -q

• TestClient httpx jaisa use hota hai: get/post + json= se body, r.status_code aur r.json() se assertions. Server chalane ki zarurat nahi — test ke andar hi request jaati hai.
• Har route ka ek happy path + ek failure path: 201 ke saath 422/404/401 bhi. Status code contract ka hissa hai, sirf body nahi.
• Failure tests documentation hain: naya banda test padh kar jaan jaata hai ki invalid input par kya hota hai.
• Dependencies ko override karke tests ko DB/tokens se azaad rakho — deterministic tests, zero network.

Exercise

Common mistakes

• async def ke andar sync client rakhna: requests.get() async endpoint mein poora event loop block karta hai — ek slow upstream sab users rok deta hai. Fix: httpx.AsyncClient + await, ya endpoint ko def rakho.
• Error ke liye 200 bhejna: Client ko body padh kar guess karna padta hai ki fail hua ya nahi, aur browsers/proxies bhi confuse hote hain. Fix: HTTPException + sahi 4xx/5xx status.
• Raw traceback client ko dikhana: Stack trace mein DB paths, env values aur internal structure chhup jaate hain. Fix: traceback server log mein, client ko detail + request id.
• Har request pe naya client ya DB connection: Connection banane ka kharcha har request par, aur exhaust hone par failures. Fix: lifespan mein ek baar banao, pool se lo, Depends se share karo.

from fastapi import FastAPI, HTTPException

app = FastAPI()

MENU = {"paneer": 220, "dal": 120}

@app.get("/menu")
def menu():
    return MENU                      # dict -> JSON automatic

@app.get("/dish/{name}")
def dish(name: str):
    if name not in MENU:
        raise HTTPException(status_code=404, detail="Nahi hai")
    return {"name": name, "price": MENU[name]}

# Chalao: uv run uvicorn main:app --reload  ->  /docs kholo