Mở đầu

Khi một request lỗi ở production, câu hỏi đầu tiên thường là: request đó tốn bao lâu? Giải pháp là viết middleware bằng BaseHTTPMiddleware: nó gắn một X-Request-ID duy nhất cho mỗi request và đo thời gian xử lý, trả về trong header X-Process-Time — áp dụng cho mọi endpoint mà không cần sửa code từng endpoint.

Vấn đề

Cách “ngây thơ” nhất là đo thời gian ngay trong từng endpoint và nhét kết quả vào body response:

import time
from fastapi import FastAPI
app = FastAPI()
@app.get("/users")
def list_users():
    start = time.perf_counter()
    users = [{"id": 1, "name": "An"}]
    return {"users": users, "process_time": f"{time.perf_counter() - start:.3f}s"}

Cách này có ba nhược điểm. Thứ nhất, bạn phải lặp lại hai dòng đo thời gian ở mọi endpoint, và endpoint nào quên thì mất số liệu. Thứ hai, định dạng không nhất quán: chỗ trả về body, chỗ trả về header, chỗ không có gì. Thứ ba, khi client báo lỗi mà không kèm mã định danh, bạn không thể đối chiếu log của request đó với log server.

Giải pháp

Middleware chạy bao quanh mọi request trước khi request tới endpoint và sau khi endpoint trả response, nên đây là đúng chỗ để gắn request-id và đo thời gian một lần duy nhất:

import time
import uuid
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
        start = time.perf_counter()
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        response.headers["X-Process-Time"] = f"{time.perf_counter() - start:.3f}s"
        return response
app.add_middleware(RequestIdMiddleware)
@app.get("/ping")
def ping():
    return {"message": "pong"}

Có ba điểm đáng chú ý. Thứ nhất, nếu client đã gửi sẵn X-Request-ID, middleware giữ nguyên mã đó thay vì sinh mã mới — nhờ vậy một request được truy vết xuyên suốt cả hệ thống. Thứ hai, time.perf_counter() đo đúng khoảng thời gian xử lý, không bị lệch khi đồng hồ hệ thống thay đổi như time.time(). Thứ ba, add_middleware áp dụng cho toàn bộ app, nên endpoint viết sau này tự động được hưởng.

Kiểm chứng nhanh bằng TestClient có sẵn của FastAPI:

from fastapi.testclient import TestClient
client = TestClient(app)
r = client.get("/ping", headers={"X-Request-ID": "req-123"})
assert r.status_code == 200
assert r.headers["X-Request-ID"] == "req-123"
assert "X-Process-Time" in r.headers
r2 = client.get("/ping")
assert r2.headers["X-Request-ID"]  # khong gui ma thi server tu sinh UUID

Áp dụng vào dự án của bạn

Hãy yêu cầu client gửi kèm X-Request-ID trong mọi lỗi họ báo, rồi grep đúng mã đó trong log server — thời gian chẩn đoán lỗi giảm từ hàng chục phút mò log xuống còn vài giây. Bạn có thể mở rộng cùng một middleware để ghi log mỗi request hoặc đếm request theo endpoint, mà vẫn chỉ sửa ở một chỗ duy nhất.

Bài viết liên quan

Câu hỏi thường gặp

Middleware khác gì dependency trong FastAPI?

Dependency chạy trong phạm vi một endpoint và thường dùng để lấy dữ liệu dùng chung như user hiện tại hay DB session. Middleware chạy bao quanh mọi request của toàn bộ app — vì vậy nó hợp với các việc xuyên suốt như request-id, đo thời gian hay ghi log.

Thứ tự add_middleware có quan trọng không?

Có. Middleware được thêm sau sẽ chạy ngoài cùng, tức là chạy trước middleware được thêm trước đó. Hãy thêm middleware đo thời gian sau cùng để nó bao được toàn bộ thời gian xử lý, kể cả thời gian xác thực.

Vì sao dùng time.perf_counter() thay vì time.time()?

time.perf_counter() dùng đồng hồ đơn điệu, không bị ảnh hưởng khi đồng hồ hệ thống bị chỉnh lại (ví dụ đồng bộ NTP). Với việc đo khoảng thời gian, nó luôn cho kết quả đúng; time.time() chỉ nên dùng khi cần mốc thời gian tuyệt đối, như ghi timestamp vào log.

Có nên dùng BaseHTTPMiddleware cho response streaming lớn không?

Với endpoint trả file hoặc stream dung lượng lớn, BaseHTTPMiddleware có thể làm giảm hiệu năng vì nó xử lý response qua lớp đệm. Trong trường hợp đó, hãy cân nhắc viết middleware ASGI thuần để toàn quyền kiểm soát luồng message.

💬 Bạn đã thử cách này trong dự án của mình chưa? Nếu gặp lỗi hoặc có cách làm gọn hơn, để lại bình luận bên dưới — mình đọc hết và sẽ bổ sung những ý hay vào bài.

Nguồn tham khảo