Sự cố đã dạy tôi về FastAPI
Đó là 2 giờ sáng thứ Ba. Flask API của chúng tôi liên tục trả về lỗi 500 trên khoảng 15% request khi chịu tải. Tôi mở log lên và vấn đề hiện ra rõ ràng: không có request validation, không có type coercion, và hoàn toàn mù quáng về dữ liệu mà client đang thực sự gửi lên. Sau bốn tiếng hotfix, tôi tự hứa sẽ làm lại đúng cách.
Cuối tuần đó, tôi chuyển các endpoint cốt lõi sang FastAPI. Không phải vì nó đang hot — mà vì nó giải quyết đúng những vấn đề tôi đang gặp phải: validation tự động, serialization, và tài liệu tích hợp sẵn mà client có thể tự dùng mà không cần gọi tôi lúc 2 giờ sáng nữa.
Đây là bài hướng dẫn tôi ước gì đã có khi tôi đang nhìn chằm chằm vào những dòng log đó. Chúng ta sẽ xây dựng một API hoàn chỉnh, đề cập đến những pattern thực sự quan trọng trong production, và bỏ qua những phần chỉ tồn tại trong các ví dụ đồ chơi.
Các khái niệm cốt lõi cần nắm trước khi viết một route
FastAPI thực sự là gì
FastAPI là một ASGI web framework được xây dựng trên Starlette để xử lý request và Pydantic để validate dữ liệu. Điểm khác biệt cốt lõi so với Flask hay Django REST Framework: validation diễn ra trước khi route function của bạn chạy. Nếu request body không khớp với schema, FastAPI tự động trả về 422 Unprocessable Entity — bạn sẽ không bao giờ nhìn thấy dữ liệu lỗi.
Trong một API tôi đã chuyển đổi, cơ chế này phát hiện nguyên một lớp lỗi ngay trong tuần đầu tiên. Những field thiếu hoặc sai định dạng vốn hay lặng lẽ qua mặt Flask và làm hỏng trạng thái phía sau đơn giản là không bao giờ vượt qua được lớp validation của FastAPI.
Pydantic Model là nền tảng
Mọi cấu trúc dữ liệu trong FastAPI đều bắt đầu từ một Pydantic BaseModel. Định nghĩa các field với type hint, và Pydantic sẽ tự động xử lý coercion và validation. Nếu client gửi "age": "25" dưới dạng string, Pydantic tự chuyển thành int. Gửi "age": "banana"? Bị từ chối trước khi code của bạn chạy.
Async theo mặc định
FastAPI hỗ trợ cả route handler đồng bộ và bất đồng bộ. Với các thao tác I/O-bound — gọi database, HTTP request đến API bên thứ ba — hãy dùng async def. Với các tác vụ CPU-bound hoặc logic đồng bộ đơn giản, def thông thường hoàn toàn ổn (FastAPI tự chạy trong thread pool).
Thực hành: Xây dựng Task Management API
Bước 1: Cài đặt Dependencies
Bắt đầu với một virtual environment sạch:
python -m venv venv
source venv/bin/activate # Trên Windows: venv\Scripts\activate
pip install fastapi uvicorn pydanticUvicorn là ASGI server. Bạn cần nó để chạy ứng dụng ở local và trong production.
Bước 2: Tạo cấu trúc ứng dụng
mkdir taskapi && cd taskapi
touch main.py models.pyBước 3: Định nghĩa Data Model
Mở models.py và định nghĩa các schema request/response:
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
class TaskCreate(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
description: Optional[str] = None
priority: int = Field(default=1, ge=1, le=5) # 1=thấp, 5=nghiêm trọng
class TaskResponse(BaseModel):
id: int
title: str
description: Optional[str]
priority: int
created_at: datetime
completed: bool = False
class Config:
from_attributes = True # Cho phép chuyển đổi ORM modelCác ràng buộc từ Field() được áp dụng tự động. Một priority có giá trị 6 sẽ bị từ chối trước khi handler của bạn chạy. Điều này đã tiết kiệm cho tôi nhiều giờ viết code phòng thủ.
Bước 4: Xây dựng API Routes
Trong main.py, kết nối các routes:
from fastapi import FastAPI, HTTPException, status
from datetime import datetime, timezone
from models import TaskCreate, TaskResponse
from typing import List
app = FastAPI(
title="Task API",
description="REST API quản lý công việc sẵn sàng cho production",
version="1.0.0"
)
# Bộ nhớ tạm cho tutorial này (thay bằng database thực tế)
db: dict = {}
counter = 0
@app.post("/tasks", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(task: TaskCreate):
global counter
counter += 1
record = {
"id": counter,
"title": task.title,
"description": task.description,
"priority": task.priority,
"created_at": datetime.now(timezone.utc),
"completed": False
}
db[counter] = record
return record
@app.get("/tasks", response_model=List[TaskResponse])
async def list_tasks():
return list(db.values())
@app.get("/tasks/{task_id}", response_model=TaskResponse)
async def get_task(task_id: int):
if task_id not in db:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Không tìm thấy task {task_id}"
)
return db[task_id]
@app.patch("/tasks/{task_id}/complete", response_model=TaskResponse)
async def complete_task(task_id: int):
if task_id not in db:
raise HTTPException(status_code=404, detail="Không tìm thấy task")
db[task_id]["completed"] = True
return db[task_id]
@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(task_id: int):
if task_id not in db:
raise HTTPException(status_code=404, detail="Không tìm thấy task")
del db[task_id]Bước 5: Chạy và kiểm tra
uvicorn main:app --reload --port 8000Trỏ trình duyệt đến http://localhost:8000/docs. FastAPI tự động tạo giao diện Swagger UI tương tác — không cần cài đặt thêm gì. Team của bạn (hoặc client đang gọi điện lúc 2 giờ sáng đó) có thể test mọi endpoint trực tiếp từ trình duyệt.
Kiểm tra bằng curl:
# Tạo một task
curl -X POST http://localhost:8000/tasks \
-H "Content-Type: application/json" \
-d '{"title": "Sửa lỗi API", "priority": 5}'
# Lấy danh sách tất cả task
curl http://localhost:8000/tasks
# Đánh dấu task hoàn thành
curl -X PATCH http://localhost:8000/tasks/1/completeBước 6: Thêm Middleware để sẵn sàng cho Production
Trước khi triển khai, thêm CORS header và một request logger cơ bản:
from fastapi.middleware.cors import CORSMiddleware
import logging
import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app.add_middleware(
CORSMiddleware,
allow_origins=["https://yourdomain.com"], # Giới hạn cụ thể trong môi trường production
allow_methods=["*"],
allow_headers=["*"],
)
@app.middleware("http")
async def log_requests(request, call_next):
start = time.time()
response = await call_next(request)
duration = round((time.time() - start) * 1000, 2)
logger.info(f"{request.method} {request.url.path} → {response.status_code} ({duration}ms)")
return responseĐó là visibility mà tôi đã thiếu lúc 2 giờ sáng. Method, path, status code, duration — bốn trường thông tin bao phủ 80% các tình huống debug production mà không cần chạm vào logic ứng dụng.
Bước 7: Chạy trên Production với Gunicorn
pip install gunicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000Flag -w 4 khởi tạo 4 tiến trình worker. Quy tắc phổ biến: 2 × số CPU core + 1. Mỗi worker xử lý request độc lập, nên một request bị treo không ảnh hưởng đến các request khác.
Bước tiếp theo
Với một deployment thực tế, đây là những gì cần bổ sung trên nền tảng này:
- Thay thế bộ nhớ dict bằng SQLAlchemy + PostgreSQL hoặc SQLModel (tác giả FastAPI tạo ra nó chính vì mục đích này)
- Thêm xác thực bằng dependency injection của FastAPI với JWT token qua
python-jose - Viết test với pytest + httpx —
TestClientcủa FastAPI giúp việc này trở nên đơn giản - Thêm rate limiting qua slowapi (phiên bản tương thích FastAPI của Flask-Limiter)
Những gì bạn có bây giờ là một hợp đồng có kiểu dữ liệu giữa client và server. Dữ liệu lỗi bị từ chối ngay tại ranh giới, tài liệu tự viết bản thân, và mọi request đều để lại dấu vết. Đó là cốt lõi. Thêm database, kết nối xác thực, và bạn có thứ gì đó có thể đưa vào tay người dùng thực.
