15 Sai Lầm Phổ Biến Khi Làm Việc Với FastAPI Trong Production
Khi viết FastAPI để chạy production, nhiều bạn cứ nghĩ “FastAPI nhanh, vậy là yên tâm rồi”. Nhưng thực tế: FastAPI chỉ nhanh khi bạn dùng đúng cách. Dùng sai một vài điểm nhỏ thôi là hiệu năng tụt thảm hại, app treo, hoặc tệ nhất là lộ bảo mật.
Dưới đây là 15 best practices quan trọng nhất. Mình giải thích chi tiết từng điều, theo đúng góc nhìn của một người đã từng deploy và vận hành FastAPI thực sự ngoài đời.
1. Không dùng async cho tác vụ blocking
Rất nhiều bạn viết FastAPI theo form:
@app.get("/data")
async def load_data():
data = open("big.csv").read()
return ...
Nhìn có vẻ “ngầu”, nhưng đây là một trong những sai lầm phổ biến nhất.
Vì sao sai?
Dù bạn dùng async, nhưng hàm open().read() là blocking I/O. Trong FastAPI, async route chạy trên event loop, nếu bạn nhét vào đó một thao tác blocking thì cả event loop bị "đóng băng".
Điều này có nghĩa:
- Nếu 1 người gọi API này → toàn bộ server chậm theo.
- Những request khác chờ hàng trăm mili giây hoặc hàng giây dù server rất rảnh.
- Hiệu năng thực tế thậm chí còn tệ hơn Flask sync.
Cách đúng
Đưa tác vụ blocking qua threadpool hoặc cứ để hàm mặc định không async. Fastapi sẽ tự nhận diện và đưa chúng vào một threed khác.
import asyncio
from concurrent.futures import ThreadPoolExecutor
from fastapi import FastAPI
app = FastAPI()
executor = ThreadPoolExecutor()
def read_big_file():
with open("big.csv") as f:
return f.read()
@app.get("/data")
async def load_data():
# chạy trong thread → không làm tắc event loop
content = await asyncio.get_event_loop().run_in_executor(
executor, read_big_file
)
return {"size": len(content)}
Tóm lại
Nếu bạn dùng async, hãy đảm bảo tất cả hàm bên trong cũng async hoặc không blocking. Nếu không chắc → để nguyên def và không async def.
2. Không dùng async nếu DB driver là sync
Ví dụ dùng SQLAlchemy sync:
# sai (vừa async, vừa query sync)
@app.get("/users/{id}")
async def get_user(id: int):
return session.get(User, id)
Vì sao sai?
session.get() là sync, FastAPI phải âm thầm chuyển nó vào threadpool mặc định → chi phí chuyển ngữ cảnh + overhead → không tăng hiệu năng mà còn chậm.
Khi nào dùng async?
- Khi bạn dùng async DB driver:
asyncpg,ormar,encode/databases,SQLAlchemy async.
Cách đúng (driver sync)
@app.get("/users/{id}")
def get_user(id: int):
return session.get(User, id)
3. Không xử lý nặng trong endpoint (CPU-heavy)
Ví dụ thật: bạn tính toán TF-IDF, chạy BERT, tách âm thanh…, nhưng làm ngay trong API.
@app.get("/process")
async def process():
return heavy_cpu_task() # sai
Vì sao sai?
- CPU nặng không liên quan async.
- Tác vụ CPU chạy trong worker hiện tại → nghẽn luôn worker → API timeout.
- Bạn tăng số worker cũng không giải quyết gốc rễ.
Cách đúng: ProcessPoolExecutor
Process → tách hẳn khỏi interpreter chính.
from concurrent.futures import ProcessPoolExecutor
worker = ProcessPoolExecutor()
@app.get("/process")
async def process():
result = await asyncio.get_event_loop().run_in_executor(
worker, heavy_cpu_task
)
return {"result": result}
4. Dùng Dependency Injection cho logic chung
Ví dụ một logic xác thực token:
@app.get("/me")
def me(request: Request):
token = request.headers["Authorization"]
# validate token thủ công – sai
Vấn đề
- Bạn phải copy-paste logic này vào tất cả route.
- Khó test.
- Khó bảo trì khi thay đổi token validation.
Cách đúng
def require_token(token: str = Header(...)):
if not validate(token):
raise HTTPException(403)
return token
@app.get("/me")
def me(token=Depends(require_token)):
return {"token": token}
Tách logic giúp code sạch, testable và ít bug.
5. Dùng BackgroundTasks cho công việc lâu
Ví dụ gửi email, ghi log, load file, gửi SMS.
Sai:
@app.post("/register")
def register(email: str):
send_email(email) # blocking
return {"ok": True}
Vì sao sai?
Người dùng chờ 2–5 giây chỉ để gửi email. Load test → API chết.
Đúng
from fastapi import BackgroundTasks
def send_email(email):
...
@app.post("/register")
def register(email: str, tasks: BackgroundTasks):
tasks.add_task(send_email, email)
return {"registered": True}
6. Không mở Swagger docs trong production
Nhiều bạn để nguyên:
/docs
/redoc
/openapi.json
Vì sao nguy hiểm?
Ai cũng truy cập được → lộ hết endpoint → dễ bị dò request.
Cách đúng
app = FastAPI(
docs_url=None,
redoc_url=None,
openapi_url=None
)
7. Tạo BaseSchema để tái dùng
Thay vì viết:
class UserOut(BaseModel):
class Config:
orm_mode = True
và lặp lại ở 10 schema khác.
Đúng
class BaseSchema(BaseModel):
class Config:
orm_mode = True
class UserOut(BaseSchema):
id: int
email: str
Giảm lặp code, dễ maintain.
8. Không build JSON thủ công — để Pydantic làm
Sai:
return {"id": user.id, "email": user.email}
Vấn đề
- Dễ thiếu field.
- Dễ sai kiểu dữ liệu.
- Không đồng nhất giữa các endpoint.
Đúng
class UserOut(BaseSchema):
id: int
email: str
@app.get("/users/{id}", response_model=UserOut)
def detail(id: int, db=Depends(get_db)):
return db.get(User, id)
Pydantic lo hết: serialize, validate, chuẩn hóa.
9. Không tự validate input — dùng Pydantic
Sai:
if len(name) < 3:
raise HTTPException(...)
Đúng
class UserCreate(BaseSchema):
name: constr(min_length=3)
email: EmailStr
Bạn viết ít code hơn, nhưng lại có validation mạnh hơn.
10. Dùng Dependency để validate với DB
Ví dụ kiểm tra user tồn tại:
def user_exists(id: int, db=Depends(get_db)):
user = db.get(User, id)
if not user:
raise HTTPException(404)
return user
@app.get("/users/{id}")
def detail(user=Depends(user_exists)):
return user
Giải thích: Tái dùng được validate trong nhiều route. Route không bị lộn xộn.
11. Không tạo DB session thủ công
Sai:
db = Session()
→ memory leak, connection leak.
Đúng
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Session được đóng đúng cách sau mỗi request.
12. Dùng lifespan để khởi chạy tài nguyên
Startup/shutdown event đang bị thay thế.
Đúng:
@asynccontextmanager
async def lifespan(app: FastAPI):
init_db()
yield
close_db()
app = FastAPI(lifespan=lifespan)
Dễ kiểm soát tài nguyên: DB, Redis, Kafka connection, v.v.
13. Không hardcode secret
Sai:
SECRET = "123"
Đúng
.env
SECRET=mysupersecretvalue
Python:
class Settings(BaseSettings):
SECRET: str
settings = Settings()
Chuyển mọi secret sang environment → an toàn hơn khi deploy.
14. Dùng structured logging
Log mất dạng là kẻ thù của production.
Đúng:
from loguru import logger
logger.add("logs/app.log", rotation="10 MB")
Loguru hỗ trợ:
- quay vòng file
- timestamp
- format nhất quán
15. Deploy đúng cách: Uvicorn + worker
Không chạy:
python main.py
Đúng:
uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4
Hoặc dùng:
- Docker
- Nginx reverse proxy
- Systemd / Supervisor
- Load balancing
FastAPI chỉ mạnh khi bạn chạy nhiều worker và cấu hình đúng.
All rights reserved