Intermediate

# main.py - Class-based dependencies for stateful resource sharing
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

class Paginator:                      # => Class-based dependency
    def __init__(self, max_limit: int = 100):
        self.max_limit = max_limit    # => Configuration stored on instance
                                      # => Injected once at dependency registration time

    def __call__(                     # => __call__ makes instance behave like function
        self,
        skip: int = 0,
        limit: int = 20,
    ) -> dict:
        if limit > self.max_limit:    # => Enforce maximum page size
            raise HTTPException(
                status_code=400,
                detail=f"limit cannot exceed {self.max_limit}",
            )
        return {"skip": skip, "limit": min(limit, self.max_limit)}
                                      # => Returns pagination dict for handler

paginate = Paginator(max_limit=50)    # => Create instance with custom configuration
                                      # => This instance is reused across requests

class RateLimiter:                    # => Stateful class tracking request counts
    def __init__(self, max_requests: int = 10):
        self.max_requests = max_requests
        self.requests: dict[str, int] = {}
                                      # => In-memory counter per client IP
                                      # => Production: use Redis for distributed limiting

    def __call__(self, request) -> None:
                                      # => request: fastapi.Request (type hint omitted for brevity)
        from fastapi import Request
        client_ip = "127.0.0.1"       # => Simplified; use request.client.host in real code
        count = self.requests.get(client_ip, 0)
        if count >= self.max_requests:
            raise HTTPException(status_code=429, detail="Too many requests")
                                      # => 429 Too Many Requests
        self.requests[client_ip] = count + 1

rate_limit = RateLimiter(max_requests=5)

@app.get("/items")
def list_items(
    pagination: Annotated[dict, Depends(paginate)],
                                      # => paginate instance is callable; Depends calls it
    _: Annotated[None, Depends(rate_limit)],
                                      # => rate_limit instance checks and increments counter
):
    return {"items": [], **pagination}

# main.py - Nested dependencies with request-scoped caching
from typing import Annotated, Generator
from fastapi import FastAPI, Depends

app = FastAPI()

# Simulated database connection class
class DBConnection:
    def __init__(self, conn_str: str):
        self.conn_str = conn_str
        self.closed = False
        print(f"[DB] Opened connection to {conn_str}")
                                      # => Printed once per request, not per dependency use

    def close(self):
        self.closed = True
        print(f"[DB] Closed connection to {conn_str}")

# Level 1 dependency: database connection
def get_db() -> Generator[DBConnection, None, None]:
                                      # => Generator dependency enables setup + teardown
    db = DBConnection("postgresql://localhost/mydb")
                                      # => Setup: runs before yielding
    try:
        yield db                      # => Yield the resource; FastAPI injects what is yielded
    finally:
        db.close()                    # => Teardown: runs after handler returns
                                      # => Guaranteed even if handler raises an exception

# Level 2 dependency: uses db connection
def get_user_repo(
    db: Annotated[DBConnection, Depends(get_db)],
                                      # => Declares dependency on get_db
                                      # => get_db is called once; result cached for this request
) -> dict:
    return {"repo": "UserRepo", "db": db.conn_str}
                                      # => UserRepo wraps the db connection

def get_order_repo(
    db: Annotated[DBConnection, Depends(get_db)],
                                      # => get_db is the SAME dependency as above
                                      # => FastAPI caches get_db result: same db object injected
                                      # => Connection opened ONCE, not twice
) -> dict:
    return {"repo": "OrderRepo", "db": db.conn_str}

# Level 3 handler: uses both repos
@app.get("/orders/user/{user_id}")
def user_orders(
    user_id: int,
    user_repo: Annotated[dict, Depends(get_user_repo)],
    order_repo: Annotated[dict, Depends(get_order_repo)],
                                      # => Both repos declared; get_db called once total
):
    return {
        "user_id": user_id,
        "user_repo": user_repo["repo"],
        "order_repo": order_repo["repo"],
        "same_db": user_repo["db"] == order_repo["db"],
                                      # => True: same connection object injected to both
    }

# main.py - OAuth2 Password flow login endpoint
from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
import hashlib

app = FastAPI()

# Simulated user database
fake_users_db = {
    "alice": {
        "username": "alice",
        "hashed_password": hashlib.sha256("secret".encode()).hexdigest(),
                                      # => In production: use bcrypt via passlib
                                      # => hashlib.sha256 is INSECURE for passwords
        "email": "alice@example.com",
        "disabled": False,
    }
}

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
                                      # => Declares that tokens come from POST /token
                                      # => Adds "Authorize" button to Swagger UI
                                      # => tokenUrl is the endpoint that issues tokens

class Token(BaseModel):               # => OAuth2 token response model
    access_token: str
    token_type: str                   # => Always "bearer" for OAuth2 Bearer tokens

def verify_password(plain: str, hashed: str) -> bool:
    return hashlib.sha256(plain.encode()).hexdigest() == hashed
                                      # => Check if plain password matches hash
                                      # => Use passlib.context.verify() in production

@app.post("/token", response_model=Token)
                                      # => Standard OAuth2 token endpoint at POST /token
async def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
                                      # => OAuth2PasswordRequestForm reads username/password
                                      # => from application/x-www-form-urlencoded body
                                      # => Fields: form_data.username, form_data.password
):
    user = fake_users_db.get(form_data.username)
    if not user or not verify_password(form_data.password, user["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
                                      # => WWW-Authenticate header required by OAuth2 spec
        )
    # In production: generate real JWT token here
    access_token = f"fake-token-for-{user['username']}"
                                      # => Real code: create_access_token(data={"sub": username})
    return Token(access_token=access_token, token_type="bearer")
                                      # => Response: {"access_token": "...", "token_type": "bearer"}

# main.py - JWT token creation and verification
# pip install "python-jose[cryptography]" passlib[bcrypt]
from datetime import datetime, timedelta, timezone
from typing import Annotated, Optional
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt           # => python-jose library for JWT operations
from pydantic import BaseModel

app = FastAPI()

SECRET_KEY = "your-256-bit-secret-key-change-in-production"
                                         # => NEVER hardcode in production; use env var
ALGORITHM = "HS256"                      # => HMAC-SHA256 signing algorithm
ACCESS_TOKEN_EXPIRE_MINUTES = 30         # => Token valid for 30 minutes

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class TokenData(BaseModel):              # => Decoded token payload model
    username: Optional[str] = None

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    to_encode = data.copy()              # => Copy to avoid mutating the original dict
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
                                         # => Default 15 minutes if not specified
    to_encode.update({"exp": expire})    # => Add expiration claim to payload
                                         # => JWT spec: "exp" is Unix timestamp
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
                                         # => Signs payload with SECRET_KEY
                                         # => Returns base64url-encoded JWT string
    return encoded_jwt                   # => "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
                                         # => oauth2_scheme extracts Bearer token from header
                                         # => Authorization: Bearer eyJhbGciO...
) -> str:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
                                         # => Verifies signature AND checks expiration
                                         # => Raises JWTError if invalid or expired
        username: Optional[str] = payload.get("sub")
                                         # => "sub" (subject) claim contains the user identifier
        if username is None:
            raise credentials_exception
        return username                  # => Return verified username string
    except JWTError:                     # => Invalid signature, expired token, malformed JWT
        raise credentials_exception

@app.get("/users/me")
async def get_me(
    current_user: Annotated[str, Depends(get_current_user)],
                                         # => get_current_user runs before handler
                                         # => Returns username if token valid, 401 if not
):
    return {"username": current_user}    # => Only reached with valid JWT

# main.py - Role-based access control with JWT claims
from typing import Annotated, List
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from pydantic import BaseModel

app = FastAPI()
SECRET_KEY = "change-me-in-production"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class User(BaseModel):
    username: str
    roles: List[str]                  # => List of role strings: ["user", "admin"]

def decode_token(token: str) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        roles = payload.get("roles", [])
                                      # => roles claim: ["user"] or ["user", "admin"]
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return User(username=username, roles=roles)
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
) -> User:
    return decode_token(token)        # => Returns User model with username and roles

def require_role(*required_roles: str):
                                      # => Factory function returning a dependency
                                      # => Call with: Depends(require_role("admin"))
    def role_checker(
        user: Annotated[User, Depends(get_current_user)],
    ) -> User:
        for role in required_roles:   # => Check if user has ANY of the required roles
            if role in user.roles:
                return user           # => Role found: allow access, return user
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
                                      # => 403 Forbidden: authenticated but not authorized
            detail=f"Requires role(s): {', '.join(required_roles)}",
        )
    return role_checker               # => Return the checker function as the dependency

@app.get("/profile")
def user_profile(
    user: Annotated[User, Depends(get_current_user)],
                                      # => Requires any authenticated user
):
    return {"username": user.username, "roles": user.roles}

@app.delete("/admin/users/{user_id}")
def admin_delete_user(
    user_id: int,
    admin: Annotated[User, Depends(require_role("admin"))],
                                      # => Requires "admin" role; 403 otherwise
):
    return {"deleted": user_id, "by": admin.username}

@app.get("/reports")
def view_reports(
    user: Annotated[User, Depends(require_role("admin", "analyst"))],
                                      # => Requires "admin" OR "analyst" role
):
    return {"reports": [], "viewer": user.username}

# database.py - Async SQLAlchemy configuration
# pip install "sqlalchemy[asyncio]" asyncpg
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from typing import AsyncGenerator

# Async database URL (note: postgresql+asyncpg://)
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
                                      # => asyncpg is the async PostgreSQL driver
                                      # => Not psycopg2 (that's blocking)

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,                     # => Number of connections to maintain in pool
    max_overflow=20,                  # => Additional connections when pool is full
    echo=False,                       # => Set True to log all SQL statements
                                      # => Useful for debugging; disable in production
)
                                      # => engine manages the connection pool lifecycle

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,             # => Use AsyncSession (not regular Session)
    expire_on_commit=False,          # => Don't expire objects after commit
                                     # => Prevents lazy-load errors after session closes
)

class Base(DeclarativeBase):         # => Declarative base for all models
    pass                             # => All models inherit from Base

# Dependency to inject database session
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
                                     # => Create session using async context manager
                                     # => Session opened before yield
        try:
            yield session            # => Inject session into route handler
        except Exception:
            await session.rollback() # => Roll back on any unhandled exception
            raise                    # => Re-raise after rollback
        finally:
            await session.close()   # => Always close session (context manager handles this)

# models.py - SQLAlchemy async model
from sqlalchemy import String, Float, Integer
from sqlalchemy.orm import Mapped, mapped_column

class Item(Base):                    # => ORM model inheriting from Base
    __tablename__ = "items"          # => Database table name

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
                                     # => Primary key with B-tree index
    name: Mapped[str] = mapped_column(String(100), nullable=False)
                                     # => VARCHAR(100) NOT NULL column
    price: Mapped[float] = mapped_column(Float, nullable=False)
    description: Mapped[str | None] = mapped_column(String(500), nullable=True)
                                     # => Nullable VARCHAR(500)

# main.py - CRUD operations with async SQLAlchemy
from typing import Annotated, List, Optional
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, delete
from pydantic import BaseModel

# Assume engine, AsyncSessionLocal, Base, get_db, Item (model) from previous example

app = FastAPI()

class ItemCreate(BaseModel):          # => Request body for creating items
    name: str
    price: float
    description: Optional[str] = None

class ItemUpdate(BaseModel):          # => Request body for updating items (all optional)
    name: Optional[str] = None
    price: Optional[float] = None
    description: Optional[str] = None

class ItemResponse(BaseModel):        # => Response model with database ID
    id: int
    name: str
    price: float
    description: Optional[str]

    class Config:
        from_attributes = True        # => Enable ORM mode: read from SQLAlchemy model attrs
                                      # => Allows Pydantic to read Item.id, Item.name etc.

@app.post("/items", response_model=ItemResponse)
async def create_item(
    item_data: ItemCreate,
    db: Annotated[AsyncSession, Depends(get_db)],
                                      # => Inject async session from get_db dependency
):
    db_item = Item(**item_data.model_dump())
                                      # => Create ORM instance from Pydantic model dict
    db.add(db_item)                   # => Stage item for insertion (not yet in DB)
    await db.commit()                 # => Execute INSERT, commit transaction
    await db.refresh(db_item)         # => Reload to get DB-generated values (id, created_at)
    return db_item                    # => Return ORM instance; response_model converts it

@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int, db: Annotated[AsyncSession, Depends(get_db)]):
    result = await db.execute(select(Item).where(Item.id == item_id))
                                      # => select(Item).where(...) builds SQL: SELECT * FROM items WHERE id = ?
    item = result.scalar_one_or_none()
                                      # => scalar_one_or_none: returns Item or None
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.get("/items", response_model=List[ItemResponse])
async def list_items(
    skip: int = 0,
    limit: int = 20,
    db: Annotated[AsyncSession, Depends(get_db)] = None,
):
    result = await db.execute(select(Item).offset(skip).limit(limit))
                                      # => .offset(skip) = SQL OFFSET; .limit(limit) = SQL LIMIT
    return result.scalars().all()     # => scalars() extracts Item objects; all() returns list

@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int, db: Annotated[AsyncSession, Depends(get_db)]):
    result = await db.execute(delete(Item).where(Item.id == item_id))
                                      # => Executes: DELETE FROM items WHERE id = ?
    if result.rowcount == 0:          # => rowcount: number of rows affected
        raise HTTPException(status_code=404, detail="Item not found")
    await db.commit()                 # => Commit the DELETE transaction

# main.py - SQLAlchemy relationships and eager loading
from typing import List
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship, selectinload
from pydantic import BaseModel
from database import Base, get_db    # => From previous example

app = FastAPI()

class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    books: Mapped[List["Book"]] = relationship(
        "Book",
        back_populates="author",      # => Bidirectional relationship
        lazy="select",                # => Default: lazy load (don't load unless accessed)
                                      # => With async, always use eager loading instead
    )

class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column()
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
                                      # => Foreign key referencing authors.id
    author: Mapped["Author"] = relationship("Author", back_populates="books")
                                      # => Many-to-one: each book has one author

class BookResponse(BaseModel):
    id: int
    title: str
    author_name: str

    class Config:
        from_attributes = True

class AuthorWithBooks(BaseModel):
    id: int
    name: str
    books: List[dict]                 # => Include related books in response

    class Config:
        from_attributes = True

@app.get("/authors/{author_id}/with-books")
async def get_author_with_books(
    author_id: int,
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Author)
        .where(Author.id == author_id)
        .options(selectinload(Author.books))
                                      # => selectinload: eager loading for async
                                      # => Issues a second SELECT for books where author_id = ?
                                      # => NOT a JOIN; avoids N+1 for large result sets
    )
    author = result.scalar_one_or_none()
    if author is None:
        return {"error": "Author not found"}
    return {
        "id": author.id,
        "name": author.name,
        "books": [{"id": b.id, "title": b.title} for b in author.books],
                                      # => author.books is already loaded (no lazy load)
    }

# main.py - Basic WebSocket echo server
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse

app = FastAPI()

# Simple HTML client for testing WebSocket without extra tools
html = """
<!DOCTYPE html>
<html>
<script>
    var ws = new WebSocket("ws://localhost:8000/ws");
    ws.onmessage = function(event) { console.log("Received:", event.data); };
    function send(msg) { ws.send(msg); }
</script>
<button onclick="send('Hello Server')">Send</button>
</html>
"""

@app.get("/")
async def home():
    return HTMLResponse(html)         # => Serve HTML client for testing

@app.websocket("/ws")                 # => WebSocket route at ws://host/ws
async def websocket_endpoint(websocket: WebSocket):
                                      # => websocket: WebSocket object for this connection
    await websocket.accept()          # => Must call accept() to establish WebSocket handshake
                                      # => Without accept(), connection is rejected
    try:
        while True:                   # => Keep connection open; loop until disconnect
            data = await websocket.receive_text()
                                      # => Blocks until client sends a text message
                                      # => data is the string the client sent
                                      # => Also: receive_bytes(), receive_json()
            print(f"Received: {data}")
            await websocket.send_text(f"Echo: {data}")
                                      # => Send response back to same client
                                      # => Also: send_bytes(), send_json()
    except WebSocketDisconnect:       # => Raised when client closes connection
        print("Client disconnected")  # => Normal disconnect; no need to re-raise
                                      # => Connection is already closed at this point

# main.py - WebSocket connection manager for broadcasting
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:              # => Manages multiple WebSocket connections
    def __init__(self):
        self.active_connections: List[WebSocket] = []
                                      # => List of all currently connected clients

    async def connect(self, websocket: WebSocket) -> None:
        await websocket.accept()      # => Accept the WebSocket handshake
        self.active_connections.append(websocket)
                                      # => Add to active connections list
        print(f"Client connected. Total: {len(self.active_connections)}")

    def disconnect(self, websocket: WebSocket) -> None:
        self.active_connections.remove(websocket)
                                      # => Remove from list on disconnect
        print(f"Client disconnected. Total: {len(self.active_connections)}")

    async def send_personal(self, message: str, websocket: WebSocket) -> None:
        await websocket.send_text(message)
                                      # => Send to one specific client only

    async def broadcast(self, message: str) -> None:
        for connection in self.active_connections:
                                      # => Send same message to ALL connected clients
            await connection.send_text(message)
                                      # => Each send is async; clients receive in order

manager = ConnectionManager()         # => Single instance shared across all connections

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
                                      # => Wait for message from this client
            await manager.send_personal(
                f"[You] {data}", websocket,
            )                         # => Echo back to sender labeled "[You]"
            await manager.broadcast(
                f"[Client {client_id}] {data}",
            )                         # => Broadcast to ALL connected clients
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"[System] Client {client_id} left the chat")
                                      # => Notify remaining clients of departure

# test_main.py - Testing FastAPI endpoints with TestClient
# pip install httpx pytest
from fastapi import FastAPI, Depends, HTTPException
from fastapi.testclient import TestClient
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    description: Optional[str] = None

items_db: dict[int, Item] = {
    1: Item(name="Widget", price=9.99),
}

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

@app.post("/items", status_code=201)
def create_item(item: Item):
    new_id = max(items_db.keys()) + 1 if items_db else 1
    items_db[new_id] = item
    return {"id": new_id, **item.model_dump()}

# --- Tests below ---

client = TestClient(app)              # => Create test client bound to app
                                      # => No real server; requests handled in-process

def test_get_existing_item():
    response = client.get("/items/1")  # => HTTP GET request to /items/1
    assert response.status_code == 200 # => Check status code
    data = response.json()             # => Parse JSON response body
    assert data["name"] == "Widget"    # => Check specific field value
    assert data["price"] == 9.99

def test_get_nonexistent_item():
    response = client.get("/items/999")
    assert response.status_code == 404  # => Expect 404 for missing item
    assert response.json()["detail"] == "Item not found"
                                        # => Check error message

def test_create_item():
    payload = {"name": "Gadget", "price": 49.99}
    response = client.post("/items", json=payload)
                                        # => json= sends Content-Type: application/json
    assert response.status_code == 201  # => Expect 201 Created
    data = response.json()
    assert data["name"] == "Gadget"
    assert "id" in data                 # => Check that ID was assigned

def test_create_item_validation_error():
    response = client.post("/items", json={"name": "Bad"})
                                        # => Missing required "price" field
    assert response.status_code == 422  # => Expect 422 Unprocessable Entity
    errors = response.json()["detail"]
    assert any(e["loc"] == ["body", "price"] for e in errors)
                                        # => Check that price field is in errors

# test_auth.py - Overriding dependencies for isolated testing
from fastapi import FastAPI, Depends, HTTPException
from fastapi.testclient import TestClient
from typing import Annotated

app = FastAPI()

# --- Application code ---

def get_current_user() -> dict:       # => Production: verifies JWT from request header
    # In production: decode JWT, verify, return user
    raise HTTPException(status_code=401, detail="Not authenticated")
                                      # => This is never called in tests (overridden)

def get_db():                         # => Production: returns real database session
    raise RuntimeError("Real DB not available in tests")
                                      # => This is never called in tests (overridden)

@app.get("/profile")
def get_profile(
    user: Annotated[dict, Depends(get_current_user)],
    db = Depends(get_db),
):
    return {"username": user["username"], "email": user["email"]}

# --- Test code ---

def fake_current_user() -> dict:      # => Test double for get_current_user
    return {"username": "testuser", "email": "test@example.com"}
                                      # => Returns hardcoded user without auth logic

class FakeDB:                         # => Test double for database session
    def query(self): return []        # => Returns empty results

def fake_get_db():                    # => Test double for get_db
    return FakeDB()                   # => Returns in-memory fake, no real DB needed

# Override dependencies for all tests using this client
app.dependency_overrides[get_current_user] = fake_current_user
                                      # => Replace get_current_user with fake_current_user
                                      # => Every Depends(get_current_user) now calls fake
app.dependency_overrides[get_db] = fake_get_db
                                      # => Replace get_db with fake_get_db

client = TestClient(app)

def test_profile_authenticated():
    response = client.get("/profile")  # => No auth header needed; fake_current_user runs
    assert response.status_code == 200
    data = response.json()
    assert data["username"] == "testuser"
                                       # => Gets testuser from fake dependency

def cleanup():
    app.dependency_overrides.clear()   # => Remove all overrides after tests
                                       # => Restore original dependencies

# main.py - Application lifecycle with lifespan context manager
from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio

# Simulated resources
class MLModel:
    def __init__(self):
        self.loaded = False
    def load(self):
        self.loaded = True
        print("[Startup] ML model loaded")
    def predict(self, data: str) -> str:
        return f"prediction for: {data}"
    def unload(self):
        self.loaded = False
        print("[Shutdown] ML model unloaded")

ml_model = MLModel()                  # => Module-level model instance

class DatabasePool:
    def __init__(self):
        self.pool = None
    async def connect(self):
        await asyncio.sleep(0.01)     # => Simulate async pool initialization
        self.pool = "connected"
        print("[Startup] Database pool initialized")
    async def close(self):
        await asyncio.sleep(0.01)     # => Simulate async pool teardown
        self.pool = None
        print("[Shutdown] Database pool closed")

db_pool = DatabasePool()

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- Startup: code runs before app accepts requests ---
    print("[Startup] Starting application...")
    ml_model.load()                   # => Sync startup task
    await db_pool.connect()           # => Async startup task
    print("[Startup] Application ready")

    yield                             # => App runs here; yields control to FastAPI

    # --- Shutdown: code runs after last request, before process exits ---
    print("[Shutdown] Shutting down...")
    await db_pool.close()             # => Async cleanup
    ml_model.unload()                 # => Sync cleanup
    print("[Shutdown] Goodbye")

app = FastAPI(lifespan=lifespan)      # => Pass lifespan to FastAPI constructor
                                      # => FastAPI calls it during startup and shutdown

@app.get("/predict")
def predict(data: str):
    if not ml_model.loaded:
        return {"error": "Model not ready"}
    return {"result": ml_model.predict(data)}
                                      # => Model guaranteed loaded because startup ran

@app.get("/db-status")
def db_status():
    return {"connected": db_pool.pool is not None}
                                      # => Pool guaranteed initialized

# config.py - Application configuration with pydantic-settings
# pip install pydantic-settings
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, AnyHttpUrl
from typing import List
from functools import lru_cache

class Settings(BaseSettings):         # => BaseSettings reads from environment variables
    model_config = SettingsConfigDict(
        env_file=".env",              # => Load from .env file if present
        env_file_encoding="utf-8",
        case_sensitive=False,         # => DB_HOST and db_host both work
    )

    # Application settings
    app_name: str = "My FastAPI App"  # => Default value; override with APP_NAME env var
    debug: bool = False               # => APP_DEBUG=true enables debug mode
    secret_key: str = Field(          # => Required: SECRET_KEY env var must be set
        ...,                          # => No default = raises error if not set
        min_length=32,                # => Must be at least 32 chars for security
    )

    # Database settings
    database_url: str = Field(        # => DATABASE_URL env var
        default="postgresql+asyncpg://user:pass@localhost/db",
    )
    db_pool_size: int = 10            # => DB_POOL_SIZE env var; default 10

    # CORS settings
    allowed_origins: List[str] = ["http://localhost:3000"]
                                      # => ALLOWED_ORIGINS env var: JSON array string
                                      # => ALLOWED_ORIGINS='["https://myapp.com"]'

    # JWT settings
    jwt_secret_key: str = Field(..., min_length=32)
                                      # => JWT_SECRET_KEY env var; required
    jwt_expire_minutes: int = 30      # => JWT_EXPIRE_MINUTES; default 30

@lru_cache()                          # => Cache Settings instance; reads env vars once
def get_settings() -> Settings:
    return Settings()                 # => Reads from environment + .env file
                                      # => lru_cache ensures single instance per process

# main.py - Using settings in application
from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

@app.get("/config-info")
def config_info(
    settings: Annotated[Settings, Depends(get_settings)],
                                      # => Depends(get_settings) injects Settings instance
                                      # => lru_cache means same Settings object every time
):
    return {
        "app_name": settings.app_name,
        "debug": settings.debug,
        "db_pool_size": settings.db_pool_size,
        # Never return secret_key or jwt_secret_key!
    }
# .env file content:
# SECRET_KEY=my-very-long-secret-key-at-least-32-chars
# JWT_SECRET_KEY=another-very-long-secret-key-here
# DATABASE_URL=postgresql+asyncpg://alice:pass@localhost/prod
# APP_NAME=Production API

# main.py - Class-based middleware with BaseHTTPMiddleware
import time
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp

app = FastAPI()

class RequestLogMiddleware(BaseHTTPMiddleware):
                                      # => Subclass BaseHTTPMiddleware for class-based middleware
    def __init__(self, app: ASGIApp, log_level: str = "INFO"):
        super().__init__(app)         # => Must call super().__init__(app)
        self.log_level = log_level    # => Store configuration on instance

    async def dispatch(self, request: Request, call_next) -> Response:
                                      # => dispatch() is called for every HTTP request
                                      # => Equivalent to the middleware function body
        start = time.perf_counter()
        client_ip = request.client.host if request.client else "unknown"
                                      # => request.client.host is the client IP address

        response = await call_next(request)
                                      # => Call the next handler (route or inner middleware)

        duration_ms = (time.perf_counter() - start) * 1000
        log_line = (
            f"[{self.log_level}] {client_ip} "
            f"{request.method} {request.url.path} "
            f"{response.status_code} {duration_ms:.1f}ms"
        )
        print(log_line)               # => "[INFO] 127.0.0.1 GET /items 200 12.3ms"
        return response

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
                                      # => Middleware that adds security headers
    async def dispatch(self, request: Request, call_next) -> Response:
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
                                      # => Prevents MIME-type sniffing attacks
        response.headers["X-Frame-Options"] = "DENY"
                                      # => Prevents clickjacking via iframes
        response.headers["X-XSS-Protection"] = "1; mode=block"
                                      # => Legacy XSS protection header
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        return response

# Add middleware in reverse order (last added runs first)
app.add_middleware(RequestLogMiddleware, log_level="INFO")
app.add_middleware(SecurityHeadersMiddleware)
                                      # => SecurityHeadersMiddleware runs first (outermost)
                                      # => RequestLogMiddleware runs second

@app.get("/items")
def list_items():
    return {"items": []}

# main.py - Streaming responses for large data and SSE
import asyncio
import json
from datetime import datetime
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def generate_large_data():      # => Async generator yielding chunks
    for i in range(100):              # => Simulate 100 database rows
        row = {"id": i, "value": f"item-{i}", "timestamp": datetime.now().isoformat()}
        yield json.dumps(row) + "\n"  # => Newline-delimited JSON (NDJSON)
                                      # => Each line is a complete JSON object
        await asyncio.sleep(0.01)     # => Simulate database row retrieval time

@app.get("/stream/data")
async def stream_data():
    return StreamingResponse(
        generate_large_data(),        # => Pass generator (not called yet)
        media_type="application/x-ndjson",
                                      # => MIME type for newline-delimited JSON
    )                                 # => FastAPI calls next() on generator per chunk
                                      # => Client receives data progressively

async def event_stream(max_events: int = 10):
                                      # => Server-Sent Events (SSE) format
    for i in range(max_events):
        event_data = json.dumps({"count": i, "time": datetime.now().isoformat()})
        yield f"data: {event_data}\n\n"
                                      # => SSE format: "data: {json}\n\n"
                                      # => Two newlines separate events
        await asyncio.sleep(1)        # => Send one event per second
    yield "data: {\"done\": true}\n\n"
                                      # => Final event signals stream end

@app.get("/stream/events")
async def stream_events():
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
                                      # => SSE media type; browser EventSource API uses this
        headers={
            "Cache-Control": "no-cache",
                                      # => Prevent caching of event stream
            "X-Accel-Buffering": "no",
                                      # => Disable nginx buffering for real-time delivery
        },
    )
# JavaScript client:
# const source = new EventSource('/stream/events');
# source.onmessage = (e) => console.log(JSON.parse(e.data));

# main.py - API versioning with mounted sub-applications
from fastapi import FastAPI
from pydantic import BaseModel

# Version 1 API
v1_app = FastAPI(
    title="My API v1",
    version="1.0.0",
    docs_url="/docs",                 # => Docs at /v1/docs
)

class ItemV1(BaseModel):
    id: int
    name: str                         # => v1: simple name field

@v1_app.get("/items/{item_id}")
def get_item_v1(item_id: int):
    return ItemV1(id=item_id, name=f"item-{item_id}")
                                      # => GET /v1/items/42 => {"id": 42, "name": "item-42"}

# Version 2 API (new fields, different behavior)
v2_app = FastAPI(
    title="My API v2",
    version="2.0.0",
    docs_url="/docs",                 # => Docs at /v2/docs
)

class ItemV2(BaseModel):
    id: int
    name: str
    display_name: str                 # => v2: additional computed field
    version: str = "v2"

@v2_app.get("/items/{item_id}")
def get_item_v2(item_id: int):
    return ItemV2(
        id=item_id,
        name=f"item-{item_id}",
        display_name=f"Item #{item_id} (v2)",
    )                                 # => GET /v2/items/42 => includes display_name

# Root app mounts versioned sub-apps
root_app = FastAPI(title="My API Gateway")

root_app.mount("/v1", v1_app)         # => All v1_app routes accessible under /v1/
                                      # => GET /v1/items/1 routes to v1_app
root_app.mount("/v2", v2_app)         # => All v2_app routes accessible under /v2/
                                      # => GET /v2/items/1 routes to v2_app

@root_app.get("/")
def api_info():
    return {
        "versions": ["v1", "v2"],
        "latest": "v2",
        "v1_docs": "/v1/docs",        # => Each version has its own docs
        "v2_docs": "/v2/docs",
    }

# routers/users.py - A complete user management router
from typing import Annotated, List
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel

# Inline dependency for demonstration
def get_current_admin():
    return {"username": "admin", "role": "admin"}
                                      # => In production: verify JWT admin role

router = APIRouter(
    prefix="/users",                  # => All routes start with /users
    tags=["users"],                   # => All routes tagged "users" in OpenAPI
    dependencies=[Depends(get_current_admin)],
                                      # => All routes in this router require admin auth
                                      # => Applied automatically without per-route declaration
    responses={
        401: {"description": "Not authenticated"},
        403: {"description": "Not authorized"},
    },                                # => These responses documented for all routes in router
)

class User(BaseModel):
    id: int
    username: str
    email: str
    active: bool = True

fake_users: List[User] = [
    User(id=1, username="alice", email="alice@example.com"),
    User(id=2, username="bob", email="bob@example.com"),
]

@router.get("/", response_model=List[User])
                                      # => Full path: GET /users/
                                      # => Requires admin (from router dependencies)
def list_users():
    return fake_users

@router.get("/{user_id}", response_model=User)
                                      # => Full path: GET /users/{user_id}
def get_user(user_id: int):
    user = next((u for u in fake_users if u.id == user_id), None)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# main.py - Include the router
from fastapi import FastAPI
# from routers.users import router as users_router  # In real project

main_app = FastAPI()
main_app.include_router(router)       # => Routes registered: GET /users/, GET /users/{user_id}
                                      # => Admin dependency applied to both

# main.py - Pydantic v2 model validators and computed fields
from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel, Field, model_validator, computed_field
import re

app = FastAPI()

class UserRegistration(BaseModel):
    username: str = Field(min_length=3, max_length=50)
    password: str = Field(min_length=8)
    confirm_password: str             # => Must match password
    email: str

    @model_validator(mode="after")    # => Runs after all individual field validators
                                      # => mode="after": receives fully validated model instance
                                      # => mode="before": receives raw input dict
    def passwords_must_match(self) -> "UserRegistration":
                                      # => self is the model instance
                                      # => Must return self (or raise ValueError)
        if self.password != self.confirm_password:
            raise ValueError("password and confirm_password must match")
                                      # => ValueError message included in 422 response
        return self

    @model_validator(mode="after")
    def username_not_in_password(self) -> "UserRegistration":
        if self.username.lower() in self.password.lower():
            raise ValueError("password must not contain username")
        return self

class Product(BaseModel):
    name: str
    price: float = Field(gt=0)
    discount_percent: float = Field(ge=0, le=100, default=0)

    @computed_field                   # => Field computed from other fields
    @property                         # => Must also be a @property
    def discounted_price(self) -> float:
        return round(self.price * (1 - self.discount_percent / 100), 2)
                                      # => price=100, discount=10 => discounted_price=90.0
                                      # => Included in JSON response automatically

    @computed_field
    @property
    def on_sale(self) -> bool:
        return self.discount_percent > 0
                                      # => True if discount_percent > 0

@app.post("/register")
def register(user: UserRegistration):
    return {"username": user.username, "email": user.email}
                                      # => confirm_password not in response (response_model could filter)

@app.post("/products")
def create_product(product: Product):
    return product                    # => Includes discounted_price and on_sale in response
                                      # => POST {"name": "Widget", "price": 100, "discount_percent": 20}
                                      # => Returns: {"name": "Widget", "price": 100,
                                      # =>           "discount_percent": 20,
                                      # =>           "discounted_price": 80.0, "on_sale": true}

# main.py - Pydantic aliases for JSON field name mapping
from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel, Field
from pydantic.alias_generators import to_camel

app = FastAPI()

class OrderItem(BaseModel):
    model_config = {                  # => Pydantic v2 model configuration
        "populate_by_name": True,     # => Accept both snake_case and camelCase
                                      # => Without this: only alias works for input
        "alias_generator": to_camel,  # => Auto-generate camelCase aliases for all fields
                                      # => snake_case Python names <-> camelCase JSON names
    }

    product_id: int                   # => Python: product_id; JSON: productId (auto-aliased)
    unit_price: float                 # => Python: unit_price; JSON: unitPrice
    quantity: int
    discount_code: Optional[str] = None
                                      # => Python: discount_code; JSON: discountCode

class LegacyItem(BaseModel):          # => Manual aliases for non-standard external field names
    item_id: int = Field(alias="ItemID")
                                      # => JSON field "ItemID" maps to Python item_id
    item_name: str = Field(alias="ItemName")
    unit_cost: float = Field(alias="Cost_USD")
                                      # => Non-standard: legacy system uses Cost_USD

@app.post("/orders/items", response_model=OrderItem)
def create_order_item(item: OrderItem):
                                      # => Accepts JSON with camelCase keys:
                                      # => {"productId": 1, "unitPrice": 9.99, "quantity": 2}
    return item                       # => Response also uses camelCase (alias_generator)

@app.post("/legacy/items", response_model=LegacyItem)
def create_legacy_item(item: LegacyItem):
                                      # => Accepts JSON: {"ItemID": 1, "ItemName": "Widget", "Cost_USD": 9.99}
                                      # => item.item_id = 1, item.item_name = "Widget"
    return item
# Note: To serialize responses using aliases, add:
# @app.post("/orders/items", response_model=OrderItem)
# def ...:
#     return item.model_dump(by_alias=True)  # => Returns {"productId": ...} instead of {"product_id": ...}

# main.py - In-process caching strategies
from functools import lru_cache
from typing import Annotated
import time
from fastapi import FastAPI, Depends

app = FastAPI()

# Process-level cache: persists for the lifetime of the process
@lru_cache(maxsize=128)               # => Cache up to 128 unique argument combinations
def expensive_computation(key: str) -> dict:
                                      # => Function must be deterministic for caching to be safe
    print(f"[CACHE MISS] Computing for key: {key}")
    time.sleep(0.1)                   # => Simulate 100ms computation
    return {"key": key, "result": key.upper() * 3, "computed_at": time.time()}
                                      # => time.time() changes each call
                                      # => But lru_cache returns CACHED result after first call
                                      # => So computed_at is the time of first computation

@lru_cache(maxsize=10)
def get_config_from_external() -> dict:
                                      # => Read config from file/env once; cache forever
    print("[CACHE MISS] Reading config")
    return {"theme": "dark", "max_items": 100}

# Request-scoped cache via dependency chain (previous Example 29 showed this)
# FastAPI caches dependency results within a single request automatically

class SimpleCache:                    # => Simple TTL-aware in-process cache
    def __init__(self, ttl_seconds: int = 60):
        self._cache: dict = {}
        self._expiry: dict = {}
        self.ttl = ttl_seconds

    def get(self, key: str):
        if key in self._cache:
            if time.time() < self._expiry[key]: # => Check if not expired
                return self._cache[key]         # => Return cached value
            del self._cache[key]                # => Remove expired entry
            del self._expiry[key]
        return None                             # => Cache miss

    def set(self, key: str, value) -> None:
        self._cache[key] = value
        self._expiry[key] = time.time() + self.ttl
                                               # => Store expiry timestamp

item_cache = SimpleCache(ttl_seconds=30)       # => Cache items for 30 seconds

@app.get("/compute/{key}")
def compute(key: str):
    return expensive_computation(key)           # => Second call with same key: instant
                                               # => lru_cache returns cached result

@app.get("/items-cached/{item_id}")
def get_item_cached(item_id: int):
    cache_key = f"item:{item_id}"
    cached = item_cache.get(cache_key)
    if cached is not None:
        return {**cached, "from_cache": True}   # => Cache hit: return immediately
    result = {"id": item_id, "name": f"item-{item_id}", "price": 9.99}
    item_cache.set(cache_key, result)           # => Store in cache for 30 seconds
    return {**result, "from_cache": False}      # => Cache miss: computed and cached