FastAPI Deep Dive Backend

Xây dựng APIs hiện đại, type-safe, production-ready với FastAPI

Learning Outcomes

Sau khi hoàn thành trang này, bạn sẽ:

• 🎯 Hiểu routing và path parameters trong FastAPI
• 🎯 Master Dependency Injection pattern cho clean architecture
• 🎯 Tích hợp Pydantic models cho request/response validation
• 🎯 Tận dụng OpenAPI generation tự động
• 🎯 Implement middleware patterns cho cross-cutting concerns
• 🎯 Tránh các Production Pitfalls phổ biến

---

Tại sao FastAPI?

# FastAPI = Modern Python + Type Hints + Async + Auto-docs
# Performance: Ngang Starlette/Uvicorn (top-tier Python frameworks)
# Developer Experience: Auto-completion, validation, documentation

from fastapi import FastAPI

app = FastAPI()

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

So sánh với Flask/Django

Feature	FastAPI	Flask	Django REST
Async native	✅	❌	⚠️ Limited
Type hints	✅ Built-in	❌ Manual	❌ Manual
Auto validation	✅ Pydantic	❌ Manual	⚠️ Serializers
Auto docs	✅ OpenAPI	❌ Manual	⚠️ drf-spectacular
Performance	🚀 Fast	🐢 Slower	🐢 Slower

---

Routing & Path Parameters

Basic Routing

from fastapi import FastAPI

app = FastAPI()

# GET request
@app.get("/users")
async def list_users():
    return [{"id": 1, "name": "HPN"}]

# POST request
@app.post("/users")
async def create_user():
    return {"id": 2, "name": "New User"}

# PUT request
@app.put("/users/{user_id}")
async def update_user(user_id: int):
    return {"id": user_id, "updated": True}

# DELETE request
@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    return {"deleted": user_id}

Path Parameters

from fastapi import FastAPI, Path

app = FastAPI()

# Basic path parameter - tự động convert type
@app.get("/users/{user_id}")
async def get_user(user_id: int):  # Auto-validated as int
    return {"user_id": user_id}

# Path với validation
@app.get("/items/{item_id}")
async def get_item(
    item_id: int = Path(
        ...,  # Required
        title="Item ID",
        description="The ID of the item to retrieve",
        ge=1,  # >= 1
        le=1000  # <= 1000
    )
):
    return {"item_id": item_id}

# Multiple path parameters
@app.get("/users/{user_id}/posts/{post_id}")
async def get_user_post(user_id: int, post_id: int):
    return {"user_id": user_id, "post_id": post_id}

Query Parameters

from fastapi import FastAPI, Query
from typing import Optional, List

app = FastAPI()

# Optional query params với defaults
@app.get("/items")
async def list_items(
    skip: int = 0,
    limit: int = 10,
    search: Optional[str] = None
):
    return {"skip": skip, "limit": limit, "search": search}

# Query với validation
@app.get("/products")
async def list_products(
    q: str = Query(
        default=None,
        min_length=3,
        max_length=50,
        regex="^[a-zA-Z0-9 ]+$",
        description="Search query"
    ),
    tags: List[str] = Query(default=[])  # Multiple values: ?tags=a&tags=b
):
    return {"q": q, "tags": tags}

Path Order Matters!

from fastapi import FastAPI

app = FastAPI()

# ⚠️ ORDER MATTERS - Specific routes BEFORE generic ones

# ✅ CORRECT ORDER
@app.get("/users/me")  # Specific route first
async def get_current_user():
    return {"user": "current"}

@app.get("/users/{user_id}")  # Generic route after
async def get_user(user_id: int):
    return {"user_id": user_id}

# ❌ WRONG ORDER - "me" would be parsed as user_id
# @app.get("/users/{user_id}")
# @app.get("/users/me")  # Never reached!

---

Request Body với Pydantic

Basic Request Body

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime

app = FastAPI()

# Pydantic model cho request body
class UserCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(default=0, ge=0, le=150)
    bio: Optional[str] = None

# Pydantic model cho response
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime

    class Config:
        from_attributes = True  # Pydantic v2 (orm_mode in v1)

@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
    # user đã được validate tự động
    return UserResponse(
        id=1,
        name=user.name,
        email=user.email,
        created_at=datetime.now()
    )

Nested Models

from pydantic import BaseModel
from typing import List, Optional

class Address(BaseModel):
    street: str
    city: str
    country: str = "Vietnam"

class UserCreate(BaseModel):
    name: str
    email: str
    addresses: List[Address] = []
    primary_address: Optional[Address] = None

@app.post("/users")
async def create_user(user: UserCreate):
    # Nested validation tự động
    return user

Request Body + Path + Query

from fastapi import FastAPI, Path, Query
from pydantic import BaseModel

class ItemUpdate(BaseModel):
    name: str
    price: float

@app.put("/items/{item_id}")
async def update_item(
    item_id: int = Path(..., ge=1),  # Path parameter
    q: str = Query(None),             # Query parameter
    item: ItemUpdate = None           # Request body
):
    return {
        "item_id": item_id,
        "q": q,
        "item": item
    }

---

Dependency Injection

Dependency Injection (DI) là pattern mạnh nhất của FastAPI - cho phép tái sử dụng logic, testing dễ dàng, và clean architecture.

Basic Dependencies

from fastapi import FastAPI, Depends

app = FastAPI()

# Simple dependency function
async def common_parameters(
    skip: int = 0,
    limit: int = 100
):
    return {"skip": skip, "limit": limit}

@app.get("/items")
async def list_items(commons: dict = Depends(common_parameters)):
    return {"params": commons}

@app.get("/users")
async def list_users(commons: dict = Depends(common_parameters)):
    return {"params": commons}  # Reuse same dependency

Class-based Dependencies

from fastapi import Depends

class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = limit

@app.get("/items")
async def list_items(pagination: Pagination = Depends()):
    # Depends() without argument uses class __init__
    return {"skip": pagination.skip, "limit": pagination.limit}

Database Session Dependency

from fastapi import Depends
from sqlalchemy.orm import Session
from typing import Generator

# Database session dependency
def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db  # Dependency với cleanup
    finally:
        db.close()

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    return user

Authentication Dependency

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)  # Your token decode logic
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

async def get_current_active_user(
    current_user: User = Depends(get_current_user)
):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

# Protected endpoint
@app.get("/users/me")
async def read_users_me(
    current_user: User = Depends(get_current_active_user)
):
    return current_user

Nested Dependencies

from fastapi import Depends

async def query_extractor(q: str = None):
    return q

async def query_or_default(q: str = Depends(query_extractor)):
    if q:
        return q
    return "default"

@app.get("/items")
async def read_items(query: str = Depends(query_or_default)):
    return {"q": query}

Dependencies với yield (Context Managers)

from fastapi import Depends
from typing import Generator

async def get_db_session() -> Generator:
    session = create_session()
    try:
        yield session
        session.commit()  # Commit nếu không có exception
    except Exception:
        session.rollback()  # Rollback nếu có exception
        raise
    finally:
        session.close()  # Cleanup luôn chạy

@app.post("/users")
async def create_user(
    user: UserCreate,
    db: Session = Depends(get_db_session)
):
    # Nếu có exception, session tự động rollback
    db_user = User(**user.dict())
    db.add(db_user)
    return db_user

---

Response Models & Status Codes

Response Model

from fastapi import FastAPI, status
from pydantic import BaseModel
from typing import List

class UserOut(BaseModel):
    id: int
    name: str
    email: str
    # Không có password field - filtered out

class UserIn(BaseModel):
    name: str
    email: str
    password: str  # Sensitive field

@app.post(
    "/users",
    response_model=UserOut,  # Filter response
    status_code=status.HTTP_201_CREATED
)
async def create_user(user: UserIn):
    # password không xuất hiện trong response
    return {"id": 1, **user.dict()}

Multiple Response Models

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Union

class UserBase(BaseModel):
    name: str
    email: str

class UserAdmin(UserBase):
    is_admin: bool = True
    permissions: List[str]

class UserRegular(UserBase):
    is_admin: bool = False

@app.get("/users/{user_id}", response_model=Union[UserAdmin, UserRegular])
async def get_user(user_id: int):
    if user_id == 1:
        return UserAdmin(name="Admin", email="admin@test.com", permissions=["all"])
    return UserRegular(name="User", email="user@test.com")

Response Model Options

@app.get(
    "/users/{user_id}",
    response_model=UserOut,
    response_model_exclude_unset=True,  # Exclude fields not explicitly set
    response_model_exclude_none=True,   # Exclude None values
    response_model_exclude={"password"},  # Exclude specific fields
    response_model_include={"id", "name"}  # Include only specific fields
)
async def get_user(user_id: int):
    return user

---

OpenAPI & Documentation

FastAPI tự động generate OpenAPI schema và interactive docs.

Customize OpenAPI

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="API for managing users and items",
    version="1.0.0",
    terms_of_service="https://example.com/terms/",
    contact={
        "name": "API Support",
        "url": "https://example.com/support",
        "email": "support@example.com",
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT",
    },
    openapi_tags=[
        {
            "name": "users",
            "description": "Operations with users",
        },
        {
            "name": "items",
            "description": "Manage items",
        },
    ]
)

Endpoint Documentation

from fastapi import FastAPI, Path, Query

@app.get(
    "/users/{user_id}",
    tags=["users"],
    summary="Get a user by ID",
    description="Retrieve a user by their unique identifier. Returns 404 if not found.",
    response_description="The user object",
    deprecated=False,
)
async def get_user(
    user_id: int = Path(
        ...,
        title="User ID",
        description="The unique identifier of the user",
        example=123
    )
):
    """
    Get a user with all their information:

    - **user_id**: unique identifier (required)

    Returns the user object with:
    - id
    - name
    - email
    - created_at
    """
    return {"user_id": user_id}

Pydantic Schema Examples

from pydantic import BaseModel, Field

class UserCreate(BaseModel):
    name: str = Field(..., example="Nguyen Van A")
    email: str = Field(..., example="nguyenvana@example.com")
    age: int = Field(default=0, example=25)

    class Config:
        json_schema_extra = {
            "example": {
                "name": "Nguyen Van A",
                "email": "nguyenvana@example.com",
                "age": 25
            }
        }

Access Docs

# Swagger UI: http://localhost:8000/docs
# ReDoc: http://localhost:8000/redoc
# OpenAPI JSON: http://localhost:8000/openapi.json

# Disable docs in production
app = FastAPI(
    docs_url=None if PRODUCTION else "/docs",
    redoc_url=None if PRODUCTION else "/redoc",
    openapi_url=None if PRODUCTION else "/openapi.json"
)

---

Middleware

Middleware xử lý requests/responses trước và sau khi đến endpoint.

Basic Middleware

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

CORS Middleware

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://frontend.example.com"],  # Specific origins
    # allow_origins=["*"],  # ⚠️ Chỉ dùng cho development
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
    expose_headers=["X-Custom-Header"],
    max_age=600,  # Cache preflight for 10 minutes
)
])

Request ID Middleware

from fastapi import FastAPI, Request
from uuid import uuid4

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid4())
    request.state.request_id = request_id
    
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

# Access in endpoint
@app.get("/")
async def root(request: Request):
    return {"request_id": request.state.request_id}

Logging Middleware

import logging
from fastapi import FastAPI, Request

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"Request: {request.method} {request.url}")
    
    response = await call_next(request)
    
    logger.info(f"Response: {response.status_code}")
    return response

Custom Middleware Class

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Skip auth for certain paths
        if request.url.path in ["/health", "/docs", "/openapi.json"]:
            return await call_next(request)
        
        # Check auth header
        auth_header = request.headers.get("Authorization")
        if not auth_header:
            return JSONResponse(
                status_code=401,
                content={"detail": "Missing authorization header"}
            )
        
        return await call_next(request)

app.add_middleware(AuthMiddleware)

---

Exception Handling

HTTPException

from fastapi import FastAPI, HTTPException, status

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = get_user_from_db(user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User {user_id} not found",
            headers={"X-Error": "User not found"}
        )
    return user

Custom Exception Handlers

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

class UserNotFoundException(Exception):
    def __init__(self, user_id: int):
        self.user_id = user_id

@app.exception_handler(UserNotFoundException)
async def user_not_found_handler(request: Request, exc: UserNotFoundException):
    return JSONResponse(
        status_code=404,
        content={
            "error": "USER_NOT_FOUND",
            "message": f"User {exc.user_id} not found",
            "request_id": getattr(request.state, "request_id", None)
        }
    )

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = get_user_from_db(user_id)
    if not user:
        raise UserNotFoundException(user_id)
    return user

Validation Error Handler

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request, 
    exc: RequestValidationError
):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })
    
    return JSONResponse(
        status_code=422,
        content={
            "error": "VALIDATION_ERROR",
            "details": errors
        }
    )

---

Background Tasks

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str, message: str):
    # Simulate slow email sending
    import time
    time.sleep(5)
    print(f"Email sent to {email}: {message}")

@app.post("/users")
async def create_user(
    email: str,
    background_tasks: BackgroundTasks
):
    # Response trả về ngay lập tức
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "User created, email will be sent"}

# Multiple background tasks
@app.post("/orders")
async def create_order(background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, "user@test.com", "Order confirmed")
    background_tasks.add_task(update_inventory, order_id=123)
    background_tasks.add_task(notify_warehouse, order_id=123)
    return {"order_id": 123}

---

Production Pitfalls

Pitfall 1: Blocking Sync Code trong Async Endpoint

import time
from fastapi import FastAPI

app = FastAPI()

# ❌ BUG: Blocking call trong async function
@app.get("/slow")
async def slow_endpoint():
    time.sleep(5)  # Blocks entire event loop!
    return {"message": "done"}

# ✅ FIX 1: Dùng asyncio.sleep cho async
import asyncio

@app.get("/slow")
async def slow_endpoint():
    await asyncio.sleep(5)  # Non-blocking
    return {"message": "done"}

# ✅ FIX 2: Dùng sync function (FastAPI tự chạy trong threadpool)
@app.get("/slow")
def slow_endpoint():  # Không có async
    time.sleep(5)  # OK - chạy trong threadpool
    return {"message": "done"}

# ✅ FIX 3: Run blocking code trong executor
from fastapi.concurrency import run_in_threadpool

@app.get("/slow")
async def slow_endpoint():
    await run_in_threadpool(time.sleep, 5)
    return {"message": "done"}

Pitfall 2: Dependency Scope Confusion

from fastapi import Depends

# ❌ BUG: Shared mutable state
class Counter:
    def __init__(self):
        self.count = 0

counter = Counter()  # Shared across all requests!

def get_counter():
    return counter

@app.get("/count")
async def increment(c: Counter = Depends(get_counter)):
    c.count += 1  # Race condition!
    return {"count": c.count}

# ✅ FIX: Create new instance per request
def get_counter():
    return Counter()  # New instance each request

Pitfall 3: Missing Response Model Validation

from pydantic import BaseModel

class UserOut(BaseModel):
    id: int
    name: str

# ❌ BUG: Response không match model
@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
    return {"id": "not-an-int", "name": 123}  # Type mismatch!
    # FastAPI sẽ raise validation error

# ✅ FIX: Ensure response matches model
@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
    return UserOut(id=user_id, name="HPN")

Pitfall 4: Không Handle Database Connection Properly

from fastapi import Depends

# ❌ BUG: Connection leak
def get_db():
    db = SessionLocal()
    return db  # Connection never closed!

# ✅ FIX: Use yield với finally
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()  # Always close

Pitfall 5: Sensitive Data trong Logs/Errors

from fastapi import HTTPException

# ❌ BUG: Expose sensitive info
@app.post("/login")
async def login(username: str, password: str):
    if not verify_password(password, user.hashed_password):
        raise HTTPException(
            status_code=401,
            detail=f"Invalid password: {password}"  # NEVER!
        )

# ✅ FIX: Generic error messages
@app.post("/login")
async def login(username: str, password: str):
    if not verify_password(password, user.hashed_password):
        raise HTTPException(
            status_code=401,
            detail="Invalid credentials"  # Generic message
        )

---

Quick Reference

from fastapi import FastAPI, Depends, HTTPException, status, Path, Query
from pydantic import BaseModel, Field

app = FastAPI()

# === ROUTING ===
@app.get("/items/{item_id}")
@app.post("/items", status_code=status.HTTP_201_CREATED)
@app.put("/items/{item_id}")
@app.delete("/items/{item_id}")

# === PATH PARAMS ===
item_id: int = Path(..., ge=1, description="Item ID")

# === QUERY PARAMS ===
q: str = Query(None, min_length=3, max_length=50)

# === REQUEST BODY ===
class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1)
    price: float = Field(..., gt=0)

# === DEPENDENCY INJECTION ===
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/items")
async def list_items(db: Session = Depends(get_db)):
    pass

# === RESPONSE MODEL ===
@app.get("/items/{item_id}", response_model=ItemOut)
async def get_item(item_id: int):
    pass

# === EXCEPTION ===
raise HTTPException(status_code=404, detail="Not found")

# === MIDDLEWARE ===
@app.middleware("http")
async def add_header(request, call_next):
    response = await call_next(request)
    return response

---

Cross-links

• Prerequisites: Asyncio, dataclasses & Pydantic
• Related: SQLAlchemy & Databases - Database integration
• See Also: API Design Patterns - REST best practices
• See Also: Production Deployment - Uvicorn, Docker

---