FastAPI là framework Python để xây dựng HTTP API, dựa trên type hints, Starlette và Pydantic. Nó phù hợp khi cần API typed, tài liệu OpenAPI tự động, validation rõ ràng, hiệu năng tốt và developer experience mạnh.

Nên dùng FastAPI cho REST API, internal services, backend cho AI/data products, service cần async I/O hoặc cần sinh SDK từ OpenAPI. Nếu app chủ yếu là server-rendered pages hoặc admin CRUD truyền thống, Django có thể phù hợp hơn.

Path operation là function gắn với một HTTP method và path, ví dụ GET /users/{id} hoặc POST /orders. FastAPI đọc type hints từ function parameters để parse path params, query params, body, header, cookie và validate tự động.

Điểm mạnh là route handler vừa là code xử lý request vừa là contract cho OpenAPI docs. Vì vậy type hints và response model nên được viết nghiêm túc, không chỉ để editor autocomplete.

Pydantic model định nghĩa schema cho request body, response, settings hoặc object trung gian. FastAPI dùng model để validate input, serialize output và sinh OpenAPI schema.

Model nên tách theo use-case: UserCreate cho input tạo mới, UserRead cho response, UserUpdate cho partial update. Không nên expose trực tiếp database model nếu nó có field nhạy cảm như password hash hoặc internal flags.

FastAPI sinh OpenAPI từ path operations, type hints, Pydantic models, status codes, dependencies và metadata. Mặc định app có Swagger UI ở /docs và ReDoc ở /redoc.

Docs tự động chỉ đáng tin khi contract được viết rõ: đặt response model, status code đúng, mô tả lỗi, phân biệt input/output schema và không trả dữ liệu tùy tiện bằng raw dict ở các endpoint quan trọng.

FastAPI phân biệt dựa trên vị trí khai báo và type: parameter có trong path template là path param; primitive parameter không nằm trong path thường là query param; Pydantic model thường là request body.

Ví dụ:

@app.post("/users/{user_id}")
async def update_user(user_id: int, notify: bool = False, payload: UserUpdate = Body()):
    return {"id": user_id, "notify": notify, "payload": payload}

Trong API public, nên đặt validation constraints rõ ràng bằng Annotated, Path, Query, Body để docs và lỗi 422 chính xác.

APIRouter gom routes theo feature/module, cho phép đặt prefix, tags, dependencies và responses chung. Root app include router qua app.include_router().

Ví dụ:

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/{user_id}")
async def get_user(user_id: int):
    return await service.get_user(user_id)

app.include_router(router)

App lớn nên tách theo domain như users, orders, billing, không tách theo technical layer kiểu tất cả controllers chung một file.

BackgroundTasks chạy task sau khi response đã gửi, phù hợp cho việc nhẹ như gửi email, ghi audit log hoặc gọi webhook không critical. Nó không thay thế queue thật.

Ví dụ:

@app.post("/signup")
async def signup(payload: Signup, tasks: BackgroundTasks):
    user = await create_user(payload)
    tasks.add_task(send_welcome_email, user.email)
    return user

Nếu task cần retry, scheduling, durability hoặc chạy lâu, nên dùng Celery/RQ/Arq, message broker hoặc workflow engine.

Lifespan quản lý resource cấp app như connection pool, cache client, ML model hoặc scheduler. Code trước yield chạy khi app startup, code sau yield chạy khi shutdown, giúp mở và đóng resource đúng vòng đời.

Ví dụ:

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = await Redis.from_url(REDIS_URL)
    yield
    await app.state.redis.aclose()

app = FastAPI(lifespan=lifespan)

Không nên tạo resource nặng ở global import nếu nó cần cleanup hoặc phụ thuộc cấu hình runtime.

FastAPI là ASGI application; ASGI là interface cho Python async web apps. Uvicorn là ASGI server phổ biến dùng để chạy FastAPI, nhận HTTP/WebSocket request và gọi app theo ASGI protocol.

Trong production, số process/workers phụ thuộc CPU, memory, blocking I/O và cách scale bằng container/orchestrator. Không tăng workers mù quáng nếu app giữ nhiều connection pool hoặc model lớn trong memory.

Status code nên phản ánh kết quả API: 200 cho đọc/update thành công, 201 cho create, 204 cho delete/no content, 400 cho input semantic sai, 401 chưa xác thực, 403 không đủ quyền, 404 không tồn tại, 409 conflict.

Ví dụ:

@app.post("/users", status_code=status.HTTP_201_CREATED, response_model=UserRead)
async def create_user(payload: UserCreate):
    return await service.create(payload)

Không nên dùng 200 cho mọi thứ vì client, monitoring và retry policy sẽ mất tín hiệu.

response_model ép output theo schema đã khai báo: validate/serialize response, lọc field không được expose và sinh OpenAPI chính xác.

Ví dụ tránh trả password hash:

class UserRead(BaseModel):
    id: int
    email: str

@app.get("/users/{user_id}", response_model=UserRead)
async def get_user(user_id: int):
    return await users_repo.get(user_id)

Không nên dựa vào việc route handler vô tình không trả field nhạy cảm; hãy để schema response làm lớp bảo vệ rõ ràng.

FastAPI dependency là callable được khai báo bằng Depends hoặc Security. FastAPI tự gọi dependency, truyền request params cần thiết, cache kết quả trong request nếu cùng dependency được dùng nhiều lần, rồi inject kết quả vào handler.

Ví dụ:

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    return await auth_service.verify(token)

@app.get("/me")
async def me(user: Annotated[User, Depends(get_current_user)]):
    return user

Dependency phù hợp cho auth, DB session, settings, pagination, tenant context và shared validation.

Dependency dùng yield phù hợp cho resource có setup/cleanup như database session, transaction, file handle hoặc external client. Code trước yield chạy trước handler, code sau yield chạy sau response path operation hoàn tất hoặc khi có exception.

Ví dụ DB session:

async def get_session():
    async with async_sessionmaker() as session:
        yield session

Không nên tạo global mutable session dùng chung nhiều request.

Mỗi request nên có session/resource scope rõ ràng để tránh leak connection và race condition.

async def chạy trong event loop và phù hợp khi handler await async I/O như async database driver, HTTP client hoặc message broker. def sync handler được chạy trong threadpool để không block event loop.

Sai lầm thường gặp là viết async def nhưng gọi thư viện blocking như requests hoặc sync DB driver bên trong; điều đó vẫn block event loop. Nếu dependency/blocking library chưa async, dùng def route hoặc chuyển sang thư viện async tương ứng.

FastAPI dùng Pydantic để validate và serialize dữ liệu, nên cần nắm các API chính của Pydantic version đang dùng trong dự án. Với Pydantic v2, các điểm hay gặp là model_config, model_validate(), model_dump() và validators bằng @field_validator/@model_validator.

Ví dụ:

class UserRead(BaseModel):
    id: int
    email: EmailStr
    model_config = ConfigDict(from_attributes=True)

Khi review FastAPI code, cần kiểm tra validators, serialization, ORM mapping và custom types vì đây là nơi dễ lệch behavior nhất.

PATCH nên dùng model riêng với fields optional, sau đó chỉ apply field nào client gửi. Với Pydantic v2, model_dump(exclude_unset=True) giúp phân biệt field không gửi với field gửi giá trị null.

Ví dụ:

class UserPatch(BaseModel):
    name: str | None = None
    bio: str | None = None

changes = payload.model_dump(exclude_unset=True)

Không nên reuse model create cho PATCH vì create thường required fields, còn PATCH phải biểu diễn partial mutation.

Lỗi validation request tự động trả 422 với chi tiết field lỗi. Business error nên dùng HTTPException hoặc custom exception + exception handler để response nhất quán.

Ví dụ:

if not user:
    raise HTTPException(status_code=404, detail="User not found")

Không nên trả { "error": ... } thủ công với status 200.

API client cần status code đúng để retry, log, alert và xử lý UX chính xác.

Pattern phổ biến: login endpoint verify credentials, cấp access token ngắn hạn; protected endpoints dùng dependency đọc Bearer token, verify signature/expiry, load user và inject current_user.

Ví dụ dependency rút gọn:

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

async def current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    return await users.get(payload["sub"])

Trong production cần refresh token, revoke/session strategy, password hashing mạnh, rotation secret/key và phân quyền theo scope/role.

CORS được cấu hình bằng CORSMiddleware. Không nên dùng wildcard origin cùng credentials trong production; hãy whitelist domain cụ thể theo môi trường.

Ví dụ:

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

CORS chỉ là chính sách browser, không phải security boundary cho API.

Backend vẫn phải authenticate và authorize request.

Mỗi request nên có session scope riêng, thường tạo bằng dependency yield. Service/repository nhận session qua dependency hoặc qua function parameter, commit/rollback ở layer rõ ràng.

Ví dụ:

async def get_db():
    async with SessionLocal() as session:
        yield session

@app.post("/users")
async def create_user(db: Annotated[AsyncSession, Depends(get_db)]):
    ...

Tránh global session dùng chung nhiều request.

Với SQLAlchemy async, cần dùng async engine/driver tương ứng như asyncpg cho PostgreSQL.

Middleware bọc toàn bộ request/response, phù hợp cho cross-cutting concern như request id, logging, timing, CORS, compression, trusted host hoặc security headers.

Ví dụ timing middleware:

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

Không nên nhét business logic vào middleware vì nó thiếu context của route handler và khó test theo domain.

Với endpoint sync/normal, dùng TestClient. Với async test cần gọi async DB/client, dùng pytest.mark.anyio và httpx.AsyncClient hoặc ASGI transport.

Ví dụ override dependency:

app.dependency_overrides[get_current_user] = lambda: User(id=1, email="a@example.com")
client = TestClient(app)
response = client.get("/me")
assert response.status_code == 200

Test tốt nên cover status code, response schema, auth path, validation lỗi 422 và side effect quan trọng.

Không nên để test dùng chung production database. Dùng test database riêng, transaction rollback, fixture tạo schema, hoặc container database tùy mức integration.

Với dependency DB session, test có thể override dependency để inject session test:

async def override_db():
    async with TestSessionLocal() as session:
        yield session

app.dependency_overrides[get_db] = override_db

Cần đảm bảo isolation giữa tests: reset data, rollback transaction hoặc tạo database/schema riêng cho từng test worker.

Docker image nên dùng base image rõ version, cài dependency có cache tốt, không chạy bằng root nếu không cần, copy source sau dependency để tận dụng layer cache, và expose healthcheck endpoint.

Ví dụ command phổ biến:

CMD ["fastapi", "run", "app/main.py", "--port", "8000"]

Trong production nên cấu hình env vars, structured logs stdout/stderr, graceful shutdown, readiness/liveness probe và giới hạn CPU/memory từ orchestrator.

StreamingResponse dùng khi response lớn hoặc sinh dần như export CSV, proxy stream, AI token stream. FileResponse dùng trả file có sẵn trên disk và xử lý headers phù hợp.

Ví dụ streaming:

async def rows():
    for row in data:
        yield json.dumps(row) + "\n"

return StreamingResponse(rows(), media_type="application/x-ndjson")

Cần chú ý client disconnect, timeout, backpressure và không giữ toàn bộ payload lớn trong memory.

SSE là stream một chiều từ server tới client qua HTTP, tự reconnect tốt trong browser và phù hợp notification, progress, AI token stream hoặc dashboard chỉ cần server push. WebSocket hai chiều và phù hợp khi client/server đều gửi liên tục.

Nếu chỉ cần server push text events, SSE đơn giản hơn WebSocket và dễ đi qua proxy hơn. Nếu cần binary, bidirectional protocol, presence hoặc low-latency interaction hai chiều, dùng WebSocket.

bytes đọc toàn bộ file vào memory, chỉ phù hợp file nhỏ. UploadFile dùng spooled file, có metadata như filename/content_type và hỗ trợ async read, phù hợp file lớn hơn.

Ví dụ:

@app.post("/avatar")
async def upload_avatar(file: UploadFile):
    content_type = file.content_type
    data = await file.read()

Production cần giới hạn size, validate content type thật, scan nếu cần, lưu object storage và tránh tin filename từ client.

Dùng Pydantic Settings hoặc pydantic-settings để parse env vars có type, default và validation. Settings nên được inject qua dependency hoặc tạo singleton cached rõ ràng.

Ví dụ:

class Settings(BaseSettings):
    database_url: str
    jwt_secret: SecretStr

@lru_cache
def get_settings() -> Settings:
    return Settings()

Không commit secret vào repo.

Tách config theo môi trường qua env vars, secret manager hoặc orchestrator secrets.

Với app nhỏ, layer đơn giản như routers/services/repositories đủ dùng. Với app lớn, nên tổ chức theo feature/domain để giảm coupling: mỗi domain có router, schemas, service, repository và tests riêng.

Quan trọng là route handler mỏng: parse request, gọi service/use-case, trả response. Business logic không nên nằm dày trong handler vì khó test và khó reuse cho worker/message consumer.

Liveness trả lời câu hỏi process còn sống không, nên đơn giản và ít dependency. Readiness trả lời app đã sẵn sàng nhận traffic chưa, có thể kiểm database/cache/critical dependency.

Ví dụ:

@app.get("/health/live")
async def live():
    return {"status": "ok"}

@app.get("/health/ready")
async def ready(db: Annotated[AsyncSession, Depends(get_db)]):
    await db.execute(text("SELECT 1"))
    return {"status": "ready"}

Nếu readiness fail, orchestrator nên ngừng route traffic nhưng không nhất thiết restart process ngay.