+3

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()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

Viblo
Hãy đăng ký một tài khoản Viblo để nhận được nhiều bài viết thú vị hơn.
Đăng kí