# FastAPI & Pydantic 

> FastAPI is the dominant Python API framework in 2026: async-first, OpenAPI auto-docs, and deeply integrated with Pydantic v2 for data validation.

---

## 1. Why FastAPI? (Exam Critical)

| Feature | FastAPI | Flask | Django |
| :--- | :--- | :--- | :--- |
| **Performance** | Near Node.js (Starlette/ASGI) | WSGI, slower | WSGI, slower |
| **Async** | Native `async/await` ✅ | Workaround only | Limited |
| **Validation** | Pydantic v2 built-in ✅ | Manual | Forms only |
| **Auto Docs** | `/docs` (Swagger) + `/redoc` ✅ | None | None |
| **Type hints** | First-class, enforced | Ignored | Ignored |
| **Best for** | APIs, ML serving, microservices | Simple apps | Full-stack web |

```
ASGI stack:  Request → Uvicorn (server) → Starlette (framework) → FastAPI (router)
WSGI stack:  Request → Gunicorn (server) → Flask/Django
```

---

## 2. Pydantic v2 : Data Validation

Pydantic validates data at runtime using Python type hints. FastAPI uses it for all request/response bodies.

### BaseModel

```python
from pydantic import BaseModel, Field, field_validator
from typing import Optional
from datetime import datetime

class User(BaseModel):
    id: int
    name: str
    email: str
    age: Optional[int] = None          # Optional with default None
    score: float = Field(default=0.0, ge=0.0, le=100.0)  # ≥0, ≤100
    created_at: datetime = Field(default_factory=datetime.utcnow)

# Instantiation : validates automatically
user = User(id=1, name="Alice", email="alice@example.com", age=30)
print(user.model_dump())  # → dict (v2 API; v1 used .dict())
print(user.model_dump_json())  # → JSON string

# Parsing from dict
user2 = User.model_validate({"id": 2, "name": "Bob", "email": "b@b.com"})
```

### Field Constraints

```python
from pydantic import Field

class Product(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)   # ... = required
    price: float = Field(..., gt=0)                         # gt=greater than
    quantity: int = Field(default=0, ge=0)                  # ge=greater or equal
    tags: list[str] = Field(default_factory=list, max_length=10)
    description: Optional[str] = Field(None, description="Product description")
```

### Custom Validators

```python
from pydantic import field_validator, model_validator

class SignupRequest(BaseModel):
    username: str
    password: str
    confirm_password: str
    email: str

    @field_validator('username')
    @classmethod
    def username_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError('username must be alphanumeric')
        return v.lower()   # transform: normalize to lowercase

    @model_validator(mode='after')   # runs after all field validators
    def passwords_match(self) -> 'SignupRequest':
        if self.password != self.confirm_password:
            raise ValueError('passwords do not match')
        return self
```

### Nested Models & Lists

```python
class Address(BaseModel):
    street: str
    city: str
    country: str = "India"

class Order(BaseModel):
    order_id: str
    items: list[str]           # list of strings
    shipping_address: Address  # nested model : validated recursively ✅
    metadata: dict[str, str] = {}

order = Order(
    order_id="ORD-001",
    items=["book", "pen"],
    shipping_address={"street": "MG Road", "city": "Bangalore"}
)
```

### Enums

```python
from enum import Enum

class Status(str, Enum):   # str Enum → serializes as string ✅
    pending = "pending"
    active = "active"
    inactive = "inactive"

class Task(BaseModel):
    title: str
    status: Status = Status.pending
```

---

## 3. FastAPI Routing : Path, Query, Body

### Minimal App

```python
from fastapi import FastAPI

app = FastAPI(title="My API", version="1.0.0")

@app.get("/")
def root():
    return {"message": "Hello World"}

# Run: uvicorn main:app --reload --port 8000
# Docs: http://localhost:8000/docs
```

### Path Parameters

```python
@app.get("/items/{item_id}")
def get_item(item_id: int):          # auto-validated as int ✅
    return {"item_id": item_id}

@app.get("/files/{file_path:path}")  # :path captures slashes too
def read_file(file_path: str):
    return {"path": file_path}
```

### Query Parameters

```python
from typing import Optional

@app.get("/search")
def search(
    q: str,                          # required query param
    page: int = 1,                   # optional with default
    size: int = 10,
    active: Optional[bool] = None    # optional, no default
):
    # GET /search?q=python&page=2&size=5
    return {"q": q, "page": page, "size": size}
```

### Request Body

```python
class CreateItem(BaseModel):
    name: str
    price: float

@app.post("/items", status_code=201)
def create_item(item: CreateItem):    # body is Pydantic model ✅
    # item.name, item.price are validated
    return {"created": item.model_dump()}
```

### Multiple Sources Together

```python
@app.put("/items/{item_id}")
def update_item(
    item_id: int,              # path param
    q: Optional[str] = None,  # query param
    item: CreateItem = None    # body (only when type is BaseModel)
):
    return {"item_id": item_id, "q": q, "item": item}
```

### Header & Cookie Parameters

```python
from fastapi import Header, Cookie

@app.get("/profile")
def profile(
    authorization: Optional[str] = Header(None),   # reads Authorization header
    session: Optional[str] = Cookie(None)          # reads session cookie
):
    return {"auth": authorization}
```

---

## 4. Response Models

```python
class UserOut(BaseModel):          # response schema (hides sensitive fields)
    id: int
    name: str
    email: str
    # no password field ✅ : never expose in response

class UserIn(BaseModel):           # input schema
    name: str
    email: str
    password: str                  # accepted on input only

@app.post("/users", response_model=UserOut, status_code=201)
def create_user(user: UserIn):
    # save user...
    return {"id": 1, "name": user.name, "email": user.email}
    # FastAPI filters response through UserOut : password stripped ✅
```

### Response Customization

```python
from fastapi.responses import JSONResponse, FileResponse, StreamingResponse
import io

@app.get("/csv")
def export_csv():
    data = "id,name\n1,Alice\n2,Bob"
    return StreamingResponse(
        io.StringIO(data),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=data.csv"}
    )
```

---

## 5. HTTP Exceptions

```python
from fastapi import HTTPException

@app.get("/items/{item_id}")
def get_item(item_id: int):
    item = db.get(item_id)
    if not item:
        raise HTTPException(
            status_code=404,
            detail=f"Item {item_id} not found"
        )
    return item

# Custom error model
from fastapi.responses import JSONResponse
from fastapi.requests import Request

@app.exception_handler(404)
async def not_found_handler(request: Request, exc: HTTPException):
    return JSONResponse(status_code=404, content={"error": exc.detail})
```

---

## 6. Dependency Injection

FastAPI's `Depends()` replaces manual middleware for reusable logic (auth, DB sessions, pagination).

```python
from fastapi import Depends

# --- Auth dependency ---
def get_current_user(token: str = Header(...)):
    user = verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

@app.get("/me")
def read_profile(user=Depends(get_current_user)):   # auto-injected ✅
    return user

# --- Pagination dependency ---
class Pagination:
    def __init__(self, page: int = 1, size: int = 10):
        self.page = page
        self.size = size
        self.offset = (page - 1) * size

@app.get("/users")
def list_users(pag: Pagination = Depends()):
    return {"offset": pag.offset, "size": pag.size}

# --- Database session dependency ---
def get_db():
    db = SessionLocal()
    try:
        yield db          # yield-based dependency = runs cleanup after request ✅
    finally:
        db.close()

@app.get("/data")
def read_data(db=Depends(get_db)):
    return db.query(Item).all()
```

---

## 7. Async Endpoints

```python
import asyncio
import httpx

# Use async when: calling other APIs, DB queries (async drivers), file I/O
@app.get("/async-fetch")
async def fetch_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
    return response.json()

# Use sync when: CPU-bound work (model inference, numpy)
@app.post("/predict")
def predict(data: InputData):
    result = heavy_model.predict(data.features)  # CPU work : sync is fine ✅
    return {"prediction": result}
```

---

## 8. Background Tasks

```python
from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    # runs after response is sent to client
    smtp.send(email, message)

@app.post("/signup")
def signup(user: UserIn, background_tasks: BackgroundTasks):
    create_user(user)
    background_tasks.add_task(send_email, user.email, "Welcome!")
    return {"status": "created"}   # returned immediately, email sent after ✅
```

---

## 9. File Uploads

```python
from fastapi import File, UploadFile

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()             # read bytes
    print(file.filename, file.content_type)  # "data.csv", "text/csv"
    
    # Save to disk
    with open(f"uploads/{file.filename}", "wb") as f:
        f.write(contents)
    return {"filename": file.filename, "size": len(contents)}

# Multiple files
@app.post("/upload-many")
async def upload_many(files: list[UploadFile] = File(...)):
    return [f.filename for f in files]
```

---

## 10. CORS Middleware

```python
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)
# For development: allow_origins=["*"] : NEVER in production ❌
```

---

## 11. Routers (Modular Apps)

```python
# routers/users.py
from fastapi import APIRouter

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/")
def list_users(): ...

@router.get("/{user_id}")
def get_user(user_id: int): ...

# main.py
from routers import users, items

app = FastAPI()
app.include_router(users.router)
app.include_router(items.router)
```

---

## 12. Testing with TestClient

```python
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_item():
    response = client.post("/items", json={"name": "Book", "price": 9.99})
    assert response.status_code == 201
    assert response.json()["created"]["name"] == "Book"

def test_not_found():
    response = client.get("/items/9999")
    assert response.status_code == 404
```

---

## 13. Lifespan Events (Startup/Shutdown)

```python
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: load model, connect DB
    app.state.model = load_model("model.pkl")
    yield
    # Shutdown: cleanup
    app.state.model = None

app = FastAPI(lifespan=lifespan)

@app.post("/predict")
def predict(data: InputData, request: Request):
    model = request.app.state.model   # access shared resource ✅
    return model.predict(data.features)
```

---

## 14. Common Patterns for ML Serving

```python
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np

app = FastAPI()
model = joblib.load("model.pkl")

class PredictRequest(BaseModel):
    features: list[float]

class PredictResponse(BaseModel):
    prediction: float
    probability: Optional[float] = None

@app.post("/predict", response_model=PredictResponse)
def predict(req: PredictRequest):
    X = np.array(req.features).reshape(1, -1)
    pred = model.predict(X)[0]
    prob = model.predict_proba(X).max() if hasattr(model, 'predict_proba') else None
    return PredictResponse(prediction=float(pred), probability=prob)
```

---

## 15. Quick Reference

```
GET    /items          → list items
GET    /items/{id}     → get one item
POST   /items          → create item (body required)
PUT    /items/{id}     → replace item (body required)
PATCH  /items/{id}     → partial update
DELETE /items/{id}     → delete item (return 204 No Content)

Auto-docs:
  http://localhost:8000/docs    → Swagger UI (interactive)
  http://localhost:8000/redoc   → ReDoc (readable)
  http://localhost:8000/openapi.json  → Raw OpenAPI schema

Run:
  uvicorn main:app --reload            # dev
  uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4  # prod

Pydantic v2 key methods:
  .model_dump()          → dict (was .dict() in v1)
  .model_dump_json()     → JSON string
  .model_validate(dict)  → parse from dict (was .parse_obj() in v1)
  .model_json_schema()   → JSON schema
```

| Pydantic Concept | Code |
| :--- | :--- |
| Required field | `name: str` or `name: str = Field(...)` |
| Optional field | `name: Optional[str] = None` |
| Constrained int | `Field(ge=0, le=100)` |
| Constrained str | `Field(min_length=1, max_length=50)` |
| Float > 0 | `Field(gt=0.0)` |
| List of models | `items: list[Item]` |
| Nested model | `address: Address` |
| Enum field | `status: StatusEnum = StatusEnum.active` |