FastAPI từ Cơ bản đến Nâng cao: Tổng hợp kiến thức cần biết

Bài viết tổng hợp kiến thức về FastAPI — framework xây dựng API bằng Python hiện đại, hiệu năng cao — theo lộ trình từ cơ bản đến nâng cao, kèm ví dụ code minh họa.

Mục lục

1. Giới thiệu chung về FastAPI
2. Cài đặt và khởi tạo dự án
3. Routing cơ bản
4. Path Parameters và Query Parameters
5. Request Body và Pydantic Model
6. Response Model
7. Validation nâng cao với Pydantic
8. Dependency Injection
9. Xử lý lỗi (Exception Handling)
10. Middleware
11. CORS
12. Xác thực và phân quyền (Authentication & Authorization)
13. Kết nối Database
14. Lập trình bất đồng bộ trong FastAPI
15. Background Tasks
16. Upload và trả về File
17. WebSocket
18. Tổ chức project lớn với APIRouter
19. Testing FastAPI
20. Tài liệu API tự động (Swagger/OpenAPI)
21. Triển khai (Deployment)
22. Tối ưu hiệu năng
23. Best Practice
24. Tài liệu tham khảo

---

1. Giới thiệu chung về FastAPI

FastAPI là một framework web hiện đại, hiệu năng cao để xây dựng API bằng Python, được tạo bởi Sebastián Ramírez, ra mắt năm 2018. Các điểm nổi bật:

• Hiệu năng cao: dựa trên Starlette (ASGI) và Pydantic, có thể sánh ngang NodeJS/Go về tốc độ.
• Type hint là trung tâm: tận dụng type hint của Python để validate dữ liệu, sinh tài liệu, và hỗ trợ autocomplete trong IDE.
• Tự động sinh tài liệu: Swagger UI (/docs) và ReDoc (/redoc) được tạo tự động từ code.
• Bất đồng bộ (async) native: hỗ trợ async/await ngay từ đầu, phù hợp xử lý nhiều request đồng thời.
• Dựa trên chuẩn mở: OpenAPI (trước là Swagger) và JSON Schema.

So với Flask, FastAPI có validation dữ liệu và tài liệu API tự động ngay từ core framework, cùng hiệu năng bất đồng bộ tốt hơn. So với Django REST Framework, FastAPI nhẹ hơn, linh hoạt hơn nhưng cần tự ghép thêm ORM, admin site nếu cần.

2. Cài đặt và khởi tạo dự án

pip install fastapi uvicorn[standard]

# main.py
from fastapi import FastAPI

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

@app.get("/")
def doc_trang_chu():
    return {"message": "Xin chào FastAPI"}

uvicorn main:app --reload

--reload giúp server tự khởi động lại khi code thay đổi, chỉ nên dùng trong môi trường phát triển.

3. Routing cơ bản

@app.get("/sinh-vien")
def lay_danh_sach_sinh_vien():
    return [{"id": 1, "ten": "An"}]

@app.post("/sinh-vien")
def tao_sinh_vien():
    return {"message": "Đã tạo sinh viên"}

@app.put("/sinh-vien/{id}")
def cap_nhat_sinh_vien(id: int):
    return {"message": f"Đã cập nhật sinh viên {id}"}

@app.delete("/sinh-vien/{id}")
def xoa_sinh_vien(id: int):
    return {"message": f"Đã xóa sinh viên {id}"}

4. Path Parameters và Query Parameters

from typing import Optional

@app.get("/sinh-vien/{mssv}")
def lay_sinh_vien(mssv: str):
    return {"mssv": mssv}

@app.get("/sinh-vien")
def tim_sinh_vien(ten: Optional[str] = None, trang: int = 1, so_luong: int = 10):
    return {"ten": ten, "trang": trang, "so_luong": so_luong}

FastAPI tự động kiểm tra kiểu dữ liệu dựa trên type hint — nếu mssv được khai báo là int mà client gửi chuỗi không phải số, FastAPI sẽ trả lỗi 422 tự động.

5. Request Body và Pydantic Model

from pydantic import BaseModel

class SinhVienCreate(BaseModel):
    ten: str
    mssv: str
    tuoi: int
    email: str | None = None

@app.post("/sinh-vien")
def tao_sinh_vien(sinh_vien: SinhVienCreate):
    return {"message": "Đã tạo", "data": sinh_vien}

Pydantic tự động validate dữ liệu đầu vào theo schema, trả về lỗi 422 với chi tiết trường nào sai nếu dữ liệu không hợp lệ.

6. Response Model

class SinhVienResponse(BaseModel):
    id: int
    ten: str
    mssv: str

    class Config:
        from_attributes = True  # (Pydantic v2) cho phép đọc từ ORM object

@app.post("/sinh-vien", response_model=SinhVienResponse)
def tao_sinh_vien(sinh_vien: SinhVienCreate):
    ...

Dùng response_model giúp:

• Lọc bỏ các trường nhạy cảm (ví dụ mật khẩu) không muốn trả về client.
• Đảm bảo response luôn đúng định dạng đã khai báo.
• Sinh tài liệu API chính xác hơn.

7. Validation nâng cao với Pydantic

from pydantic import BaseModel, Field, EmailStr, field_validator

class DangKyTaiKhoan(BaseModel):
    ten_dang_nhap: str = Field(min_length=3, max_length=20)
    email: EmailStr
    mat_khau: str = Field(min_length=8)
    tuoi: int = Field(gt=0, le=120)

    @field_validator("mat_khau")
    @classmethod
    def kiem_tra_mat_khau(cls, v):
        if not any(c.isdigit() for c in v):
            raise ValueError("Mật khẩu phải chứa ít nhất 1 chữ số")
        return v

8. Dependency Injection

Dependency Injection là một trong những điểm mạnh nhất của FastAPI, giúp tái sử dụng logic (xác thực, kết nối DB, phân trang...) một cách gọn gàng.

from fastapi import Depends

def phan_trang(trang: int = 1, so_luong: int = 10):
    return {"trang": trang, "so_luong": so_luong}

@app.get("/sinh-vien")
def lay_danh_sach(phan_trang: dict = Depends(phan_trang)):
    return {"phan_trang": phan_trang}

Dependency dạng class

class QueryParams:
    def __init__(self, tu_khoa: str = "", trang: int = 1):
        self.tu_khoa = tu_khoa
        self.trang = trang

@app.get("/tim-kiem")
def tim_kiem(params: QueryParams = Depends()):
    return {"tu_khoa": params.tu_khoa}

9. Xử lý lỗi (Exception Handling)

from fastapi import HTTPException, status

@app.get("/sinh-vien/{mssv}")
def lay_sinh_vien(mssv: str):
    sinh_vien = tim_theo_mssv(mssv)
    if not sinh_vien:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Không tìm thấy sinh viên"
        )
    return sinh_vien

Custom Exception Handler

from fastapi.requests import Request
from fastapi.responses import JSONResponse

class KhongDuQuyenException(Exception):
    def __init__(self, message: str):
        self.message = message

@app.exception_handler(KhongDuQuyenException)
def xu_ly_khong_du_quyen(request: Request, exc: KhongDuQuyenException):
    return JSONResponse(
        status_code=403,
        content={"error": exc.message}
    )

10. Middleware

import time
from fastapi import Request

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

11. CORS

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # domain frontend được phép gọi API
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

12. Xác thực và phân quyền (Authentication & Authorization)

OAuth2 với JWT (phổ biến nhất)

from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from datetime import datetime, timedelta

SECRET_KEY = "chuoi-bi-mat-cua-ban"
ALGORITHM = "HS256"

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def tao_access_token(data: dict, thoi_han: timedelta = timedelta(minutes=30)):
    to_encode = data.copy()
    to_encode.update({"exp": datetime.utcnow() + thoi_han})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def lay_user_hien_tai(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Token không hợp lệ")
        return username
    except JWTError:
        raise HTTPException(status_code=401, detail="Token không hợp lệ")

@app.get("/thong-tin-ca-nhan")
def thong_tin_ca_nhan(user: str = Depends(lay_user_hien_tai)):
    return {"username": user}

Phân quyền theo vai trò (Role-based)

def yeu_cau_vai_tro(vai_tro_yeu_cau: str):
    def kiem_tra(user: dict = Depends(lay_user_hien_tai)):
        if user["vai_tro"] != vai_tro_yeu_cau:
            raise HTTPException(status_code=403, detail="Không có quyền truy cập")
        return user
    return kiem_tra

@app.delete("/sinh-vien/{id}")
def xoa_sinh_vien(id: int, user: dict = Depends(yeu_cau_vai_tro("admin"))):
    ...

13. Kết nối Database

Với SQLAlchemy (ORM phổ biến nhất)

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "postgresql://user:password@localhost/mydb"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class SinhVien(Base):
    __tablename__ = "sinh_vien"
    id = Column(Integer, primary_key=True, index=True)
    ten = Column(String)
    mssv = Column(String, unique=True)

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

@app.get("/sinh-vien")
def lay_danh_sach(db=Depends(get_db)):
    return db.query(SinhVien).all()

Với SQLModel (kết hợp Pydantic + SQLAlchemy, cùng tác giả FastAPI)

from sqlmodel import SQLModel, Field, create_engine, Session

class SinhVien(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    ten: str
    mssv: str

engine = create_engine("sqlite:///database.db")

def get_session():
    with Session(engine) as session:
        yield session

SQLModel giúp giảm trùng lặp code vì không cần định nghĩa 2 lần (Pydantic model cho API và ORM model cho database) như khi dùng SQLAlchemy thuần.

14. Lập trình bất đồng bộ trong FastAPI

import httpx

@app.get("/goi-api-ngoai")
async def goi_api_ngoai():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

Lưu ý quan trọng: chỉ dùng async def khi các thao tác bên trong thực sự là non-blocking (dùng thư viện hỗ trợ async như httpx, asyncpg). Nếu gọi thư viện đồng bộ (blocking) như requests hay các thao tác CPU nặng bên trong hàm async def, sẽ chặn toàn bộ event loop và làm giảm hiệu năng thay vì tăng — trường hợp này nên dùng def thường, FastAPI sẽ tự chạy nó trong threadpool riêng.

15. Background Tasks

from fastapi import BackgroundTasks

def gui_email_thong_bao(email: str, noi_dung: str):
    # Giả lập gửi email tốn thời gian
    print(f"Đã gửi email tới {email}: {noi_dung}")

@app.post("/dang-ky")
def dang_ky(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(gui_email_thong_bao, email, "Chào mừng bạn!")
    return {"message": "Đăng ký thành công, email sẽ được gửi sau"}

BackgroundTasks phù hợp cho tác vụ nhẹ, chạy nhanh sau khi trả response. Với tác vụ nặng, dài hạn, nên dùng hàng đợi tác vụ chuyên dụng như Celery hoặc ARQ.

16. Upload và trả về File

from fastapi import UploadFile, File
from fastapi.responses import FileResponse

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    noi_dung = await file.read()
    with open(f"uploads/{file.filename}", "wb") as f:
        f.write(noi_dung)
    return {"filename": file.filename}

@app.get("/download/{ten_file}")
def download_file(ten_file: str):
    return FileResponse(f"uploads/{ten_file}")

17. WebSocket

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Server nhận được: {data}")
    except WebSocketDisconnect:
        print("Client đã ngắt kết nối")

18. Tổ chức project lớn với APIRouter

app/
├── main.py
├── routers/
│   ├── __init__.py
│   ├── sinh_vien.py
│   └── gia_su.py
├── models/
├── schemas/
└── services/

# routers/sinh_vien.py
from fastapi import APIRouter

router = APIRouter(prefix="/sinh-vien", tags=["Sinh viên"])

@router.get("/")
def lay_danh_sach():
    return []

# main.py
from routers import sinh_vien

app.include_router(sinh_vien.router)

Cách tổ chức này giúp project mở rộng dễ dàng, mỗi domain/module có router, schema, service riêng biệt, thuận tiện khi nhiều người cùng phát triển.

19. Testing FastAPI

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_doc_trang_chu():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Xin chào FastAPI"}

def test_tao_sinh_vien():
    response = client.post("/sinh-vien", json={
        "ten": "An", "mssv": "SV001", "tuoi": 20
    })
    assert response.status_code == 200

Với các route bất đồng bộ hoặc cần test tương tác thực sự với event loop, có thể dùng httpx.AsyncClient kết hợp pytest-asyncio.

20. Tài liệu API tự động (Swagger/OpenAPI)

FastAPI tự động sinh:

• Swagger UI: truy cập tại /docs
• ReDoc: truy cập tại /redoc
• OpenAPI schema JSON: tại /openapi.json

@app.get(
    "/sinh-vien/{mssv}",
    summary="Lấy thông tin sinh viên theo MSSV",
    description="Trả về thông tin chi tiết của một sinh viên dựa trên mã số sinh viên",
    response_description="Thông tin sinh viên",
)
def lay_sinh_vien(mssv: str):
    ...

Thêm docstring, summary, description, và ví dụ mẫu (Field(example=...) hoặc model_config) trong Pydantic model giúp tài liệu Swagger chi tiết và dễ dùng hơn cho các đội frontend/QA.

21. Triển khai (Deployment)

# Chạy với Uvicorn (production, không dùng --reload)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

# Hoặc dùng Gunicorn quản lý nhiều worker Uvicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

Dockerfile mẫu

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Trong môi trường production nên đặt reverse proxy (Nginx) phía trước, cấu hình HTTPS, và không bật --reload.

22. Tối ưu hiệu năng

• Dùng thư viện async (asyncpg, httpx, motor) cho các thao tác I/O để tận dụng tối đa lợi thế bất đồng bộ.
• Dùng connection pooling cho database thay vì mở/đóng kết nối liên tục.
• Bật caching (Redis) cho các endpoint đọc dữ liệu ít thay đổi.
• Dùng response_model_exclude_unset hoặc response_model_exclude_none để giảm dung lượng response khi cần.
• Sử dụng Pydantic v2 (nhanh hơn v1 đáng kể nhờ core viết bằng Rust — pydantic-core).
• Chạy nhiều worker process (qua Gunicorn/Uvicorn) để tận dụng đa nhân CPU, vì một process Python bị giới hạn bởi GIL.

23. Best Practice

• Tách biệt rõ ràng routers / schemas / models / services thay vì viết hết logic trong file route.
• Luôn dùng Pydantic model cho cả input (request) và output (response), tránh trả dict tùy tiện.
• Không đặt logic nghiệp vụ trong hàm route — nên đẩy xuống tầng service/repository.
• Dùng Dependency Injection cho các thành phần dùng chung (DB session, current user, phân trang).
• Quản lý cấu hình (secret key, DB URL...) bằng pydantic-settings và biến môi trường (.env), không hard-code.
• Viết unit test cho từng endpoint và integration test cho luồng nghiệp vụ chính.
• Version hóa API (/api/v1/...) ngay từ đầu để dễ nâng cấp về sau mà không phá vỡ client cũ.

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    secret_key: str

    class Config:
        env_file = ".env"

settings = Settings()

24. Tài liệu tham khảo

• FastAPI Official Documentation
• Pydantic Documentation
• Starlette Documentation
• SQLModel Documentation

---

Bài viết được tổng hợp nhằm mục đích chia sẻ kiến thức, phù hợp làm tài liệu ôn tập hoặc tra cứu nhanh khi xây dựng API với FastAPI.