FastAPI Từng Lớp Một: Cấu Trúc Sản Xuất Giúp Kiểm Thử Trở Nên Dễ Dàng

Mọi chuyện bắt đầu từ mong muốn xây dựng một ứng dụng tích hợp LLM thực thụ, buộc tác giả phải quay lại tìm hiểu các nguyên tắc cơ bản về cấu trúc dự án. Bài viết này chia sẻ kiến trúc FastAPI phân tầng hiệu quả, giúp tách biệt rõ ràng giữa xử lý HTTP, logic nghiệp vụ và cơ sở dữ liệu để tối ưu hóa khả năng kiểm thử.

Trong một thời gian dài, các dự án API của tôi thường bắt đầu rất gọn gàng nhưng sau đó âm thầm tan rã. Chúng không bị lỗi hay hỏng hóc — chúng vẫn chạy tốt. Tuy nhiên, sau vài tuần thêm tính năng, mỗi khi mở codebase ra, tôi lại dành nhiều thời gian hơn để tìm xem "mảnh code nào nằm ở đâu" thay vì thực sự viết code mới. Logic nghiệp vụ tràn lan vào các route handler, các lời gọi cơ sở dữ liệu vứt vội khắp nơi, và tệp utils.py trở thành "nghĩa địa" cho những thứ tôi không biết đặt vào đâu.

Tôi liên tục nghe khuyên rằng "hãy cấu trúc dự án của bạn đúng cách", nhưng ít ai giải thích tại sao một cấu trúc lại hoạt động — họ chỉ cho tôi biết nên tạo những thư mục nào. Vì vậy, tôi đã tạm dừng phần LLM và quay lại với những điều cơ bản.

Bài viết này là những gì tôi đúc kết được sau quá trình đó. Không phải là cẩm nang tham khảo — mà là những gì tôi đã tìm ra, được giải thích theo cách mà tôi ước có ai đó đã từng giải thích cho mình.

Cấu trúc cơ bản

Trước hết, hãy nhìn vào bố cục tổng thể của dự án:

app/
├── api/           # Ranh giới HTTP — routes, hình dạng request/response
├── core/
│   └── settings.py  # Cấu hình ứng dụng, tải một lần khi khởi động
├── services/      # Logic nghiệp vụ
├── repositories/  # Mọi thứ liên quan đến cơ sở dữ liệu
├── clients/       # Wrapper cho API bên thứ ba (Stripe, SendGrid, v.v.)
├── dependencies.py # Nối dây FastAPI Depends() — auth, user hiện tại, v.v.
└── database.py    # Quản lý pool kết nối
main.py            # App factory, lifespan, đăng ký router

Điều khiến tôi thực sự hiểu ra vấn đề là nghĩ về từng lớp như có một nhiệm vụ riêng — và quan trọng hơn là những thứ nó không được phép làm. Các thư mục gần như chỉ là thứ yếu so với tư duy đó.

Cấu hình (Settings): Đừng phân tán os.getenv() khắp nơi

Điều đầu tiên tôi nhận ra là việc gọi os.getenv("DATABASE_URL") ở ba file khác nhau là một vấn đề dai dẳng. Nếu tên biến thay đổi, hoặc bạn muốn thêm giá trị mặc định, hoặc muốn hiểu ứng dụng cần cấu hình gì — bạn sẽ phải đi săn tìm.

Giải pháp là một file duy nhất app/core/settings.py quản lý tất cả:

from pydantic_settings import BaseSettings, SettingsConfigDict

class AppSettings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
        env_prefix="APP_",
        extra="forbid",
        frozen=True,
    )

    database_url: str
    debug: bool = False
    max_connections: int = 10

Điều thực sự thuyết phục tôi ở đây là extra="forbid". Nếu bạn có lỗi chính tả trong file .env — ví dụ APP_DATABSE_URL thay vì APP_DATABASE_URL — ứng dụng sẽ sập ngay lập tức khi khởi động với lỗi xác thực. Trước khi biết điều này, tôi từng dành ages để gỡ lỗi một sự cố âm thầm do lỗi chính tả biến môi trường gây ra. Giờ thì nó sẽ báo cho bạn ngay.

frozen=True là một tùy chọn khác tôi ban đầu chưa đánh giá cao. Nó làm cho đối tượng cấu hình bất biến (immutable) sau khi được tạo — nếu bất kỳ code nào cố thay đổi cài đặt sau khi khởi động, nó sẽ ngay lập tức nhận TypeError. Ý tưởng rất đơn giản: cấu hình nên được đọc một lần khi ứng dụng khởi động, và không có gì được phép thay đổi nó giữa chừng. Nếu không có cái này, một số code có thể vô tình thay đổi đối tượng cấu hình được chia sẻ và âm thầm thay đổi hành vi cho mọi người gọi tiếp theo.

Để đảm bảo cấu hình chỉ được phân tích cú pháp một lần duy nhất, bạn hãy bọc quá trình khởi tạo bằng lru_cache:

from functools import lru_cache

@lru_cache(maxsize=1)
def get_settings() -> AppSettings:
    return AppSettings()

lru_cache là một decorator của thư viện chuẩn lưu trữ giá trị trả về của hàm. maxsize=1 nghĩa là nó lưu đúng một kết quả — thể hiện của cài đặt — và trả về cùng một đối tượng đó trong mọi lần gọi tiếp theo mà không cần chạm vào hệ thống file tệp tin nữa. File .env chỉ được đọc đúng một lần. Trong các bài kiểm tra (test), bạn gọi get_settings.cache_clear() để xóa thể hiện đã lưu và ép buộc đọc lại với các giá trị môi trường khác.

Routes: Công việc hẹp nhất trong ứng dụng

Tôi từng đặt rất nhiều logic vào các route handler. Cảm giác rất tự nhiên — route là nơi request đến, vì sao không xử lý mọi thứ ở đó?

Vấn đề là ngay khi bạn làm thế, bạn không thể tái sử dụng bất kỳ logic nào. Và rất khó để kiểm tra (test) nếu không khởi chạy một HTTP client hoàn chỉnh.

Điều đã thay đổi tư duy của tôi: công việc duy nhất của một route handler là nói chuyện HTTP. Nhận request, xác thực đầu vào, gọi thứ gì đó khác, trả về response với mã trạng thái thích hợp. Chỉ thế thôi.

@router.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(
    payload: CreateUserRequest,
    user_service: UserService = Depends(get_user_service),
):
    user = await user_service.create(payload)
    return UserResponse.model_validate(user)

Depends(get_user_service) là dependency injection của FastAPI. Khi request đến, FastAPI gọi get_user_service và tiêm kết quả vào handler của bạn. Đây là cách bạn tránh xây dựng thủ công các services và repositories bên trong mỗi route — bạn khai báo những gì bạn cần, và FastAPI giải quyết nó. Cơ chế này cũng xử lý xác thực, session cơ sở dữ liệu, giới hạn tốc độ và nhiều hơn nữa. Đây là một trong những điều làm cho FastAPI rất dễ ghép nối.

UserResponse.model_validate(user) kiểm soát rõ ràng những gì được gửi lại cho người gọi. Ngay cả khi mô hình User bên trong có các trường như hashed_password hoặc internal_flags, chúng chỉ rời khỏi máy chủ nếu UserResponse bao gồm chúng. Điều này là có chủ đích — bạn không bao giờ muốn vô tình serialize sai các trường vì quên loại trừ điều gì đó.

Mọi dự án tôi xây dựng hiện nay đều có hai endpoint kiểm tra sức khỏe (health endpoints):

@router.get("/health")
async def get_health():
    return {"status": "healthy", "version": "1.0.0"}

@router.get("/health/db")
async def get_health_db(db: Database = Depends(get_db)):
    try:
        await db.check_connection()
        return {"status": "healthy", "dependencies": {"database": "connected"}}
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail={"status": "unhealthy", "error": str(e)},
        )

/health là để kiểm tra "quá trình có đang chạy không". /health/db là để kiểm tra "quá trình có thực sự hữu ích ngay lúc này không". Tôi đã không hiểu tại sao cần cả hai cho đến khi tôi bắt đầu triển khai lên môi trường mà máy chủ khởi động rất tốt nhưng âm thầm không thể kết nối được cơ sở dữ liệu. Endpoint thứ hai sẽ phát hiện điều đó.

Services: Nơi logic thực sự nằm

Đây là lớp khiến mọi thứ khác trở nên hợp lý khi tôi cuối cùng cũng hiểu được nó.

Một service chỉ đơn giản là một lớp (class) làm một việc gì đó. Nó không biết nó đang được gọi qua HTTP. Không có đối tượng Request, không có mã trạng thái, không có serialization. Chỉ là: nhận dữ liệu sạch, làm gì đó với nó, trả về kết quả hoặc ném ra một ngoại lệ có ý nghĩa.

class UserService:
    def __init__(self, user_repo: UserRepository):
        self.user_repo = user_repo

    async def create(self, payload: CreateUserRequest) -> User:
        existing = await self.user_repo.find_by_email(payload.email)
        if existing:
            raise EmailAlreadyExistsError(payload.email)

        hashed_pw = hash_password(payload.password)
        return await self.user_repo.create(payload.email, hashed_pw)

Bởi vì nó không biết về HTTP, việc kiểm tra nó trở nên vô cùng đơn giản. Bạn chỉ cần truyền dữ liệu vào và xác nhận đầu ra. Không cần test client, không cần mock request. Chỉ riêng điều đó đã khiến tôi muốn giữ mọi thứ theo cách này.

Một điều khác tôi bắt đầu đánh giá cao: cùng một service này có thể được gọi từ một công việc nền (background job), lệnh CLI hoặc một nhiệm vụ theo lịch. Logic chỉ tồn tại ở một nơi.

Repositories: Nơi duy nhất thuộc về SQL

Trước khi tôi hiểu mô hình này, các truy vấn SQL của tôi nằm rải rác khắp nơi. Trong services, trong route handlers, đôi khi là cả hai. Nó khiến bất kỳ thay đổi lược đồ nào cũng trở thành một cuộc đi săn manh mún.

Ý tưởng rất đơn giản: chỉ có một lớp được phép chạm vào cơ sở dữ liệu, và mọi thứ khác phải đi qua nó.

class UserRepository:
    def __init__(self, db: Database):
        self.db = db

    async def find_by_email(self, email: str) -> Optional[User]:
        async with self.db.pool.connection() as conn:
            async with conn.cursor() as cur:
                await cur.execute(
                    "SELECT id, email, created_at FROM users WHERE email = %s",
                    (email,)
                )
                row = await cur.fetchone()
                return User.model_validate(dict(row)) if row else None

Lợi ích thực tế tôi cảm nhận được: khi một bảng bị đổi tên hoặc một cột thay đổi, tôi chỉ cần đi đến một file duy nhất. Service không quan tâm — nó vẫn gọi find_by_email và nhận về một User. Đối với các miền phức tạp hơn, bạn có thể tách repositories hơn nữa hoặc giới thiệu các mô hình khác — nhưng đó là một vấn đề đáng để giải quyết sau, khi ứng dụng của bạn thực sự lớn mạnh.

Lớp Database: Quản lý Pool kết nối

Tôi không thực sự hiểu về connection pools khi mới bắt đầu. Tôi ngây thơ mở một kết nối mới cho mọi request, điều này hoạt động tốt ở địa phương và bắt đầu sụp đổ dưới tải thực.

Một connection pool giữ một số lượng kết nối cơ sở dữ liệu mở và sẵn sàng (warm) để các request có thể mượn một cái, sử dụng nó, và trả lại mà không có chi phí của một cuộc bắt tay kết nối đầy đủ mỗi lần.

from psycopg_pool import AsyncConnectionPool
from typing import Optional

class Database:
    def __init__(self):
        self._pool: Optional[AsyncConnectionPool] = None

    async def connect(self):
        settings = get_settings()
        self._pool = AsyncConnectionPool(settings.database_url, open=False)
        await self._pool.open()

    async def disconnect(self):
        if self._pool:
            await self._pool.close()
            self._pool = None

    @property
    def pool(self) -> AsyncConnectionPool:
        if self._pool is None:
            raise RuntimeError("Database not connected. Call connect() first.")
        return self._pool

    async def check_connection(self):
        async with self.pool.connection() as conn:
            async with conn.cursor() as cur:
                await cur.execute("SELECT 1")

db = Database()

Đối số open=False là thứ tôi từng phải tra cứu. Nó bảo pool không mở bất kỳ kết nối nào tại thời điểm khởi tạo — bạn kiểm soát khi nào nó mở bằng cách gọi await self._pool.open() một cách rõ ràng, bên trong lifespan. Như vậy, các kết nối không được tạo tại thời điểm import trước khi ứng dụng sẵn sàng.

Thuộc tính pool với bộ bảo vệ RuntimeError là thứ tôi thêm vào sau khi gặp một lỗi gây bối rối khi có thứ gì đó cố sử dụng pool quá sớm và tôi nhận được lỗi NoneType sâu trong driver. Bộ bảo vệ làm cho sai lầm trở nên rõ ràng ngay lập tức.

main.py: Chỉ là việc nối dây

Khi mọi thứ khác đã tại vị, main.py trở nên gần như nhàm chán — điều mà giờ tôi nghĩ là một dấu hiệu tốt.

from contextlib import asynccontextmanager
from app.database import db
from fastapi import FastAPI
from app.api.routes import router

@asynccontextmanager
async def lifespan(app: FastAPI):
    await db.connect()
    yield
    await db.disconnect()

app = FastAPI(lifespan=lifespan)
app.include_router(router)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Trình quản lý ngữ cảnh lifespan là nơi logic khởi động và tắt đi. Mọi thứ trước yield chạy khi ứng dụng khởi động. Mọi thứ sau đó chạy khi nó tắt. Việc mở pool cơ sở dữ liệu ở đây cảm giác là đúng đắn một khi tôi hiểu mô hình — nó rõ ràng, có thứ tự, và bạn biết chính xác khi nào các kết nối tồn tại và khi nào thì không.

Kiểm thử (Testing): Phần mất nhiều thời gian nhất để hiểu

Tôi muốn dành nhiều thời gian hơn cho phần này so với các phần khác, bởi vì tôi nghĩ kiểm thử vừa là điều quan trọng nhất cần làm đúng, vừa là điều khó nhất để thực sự nắm bắt — không phải về mặt kỹ thuật, mà là tư duy đằng sau nó.

Cấu trúc tôi mô tả ở trên được xây dựng, phần lớn, để mọi thứ có thể được kiểm tra một cách cô lập. Tất cả các lớp với công việc hẹp — chúng tồn tại một phần là để bạn có thể hoán đổi cơ sở dữ liệu thực bằng một cái giả trong các bài kiểm tra mà ứng dụng không nhận ra. Việc hiểu điều đó đã thay đổi cách tôi nhìn toàn bộ kiến trúc.

Có ba loại kiểm tra và chúng làm những việc khác nhau.

• Unit tests (Kiểm thử đơn vị): Chạm vào một service hoặc utility trong sự cô lập hoàn toàn. Không có cơ sở dữ liệu, không có HTTP. Bạn xây dựng một service, truyền vào các dependencies giả, gọi một phương thức, xác nhận kết quả. Nhanh, chính xác, và chúng cho bạn biết chính xác cái gì bị hỏng.
• Integration tests (Kiểm thử tích hợp): Chạm vào một route qua HTTP với một cơ sở dữ liệu (thử nghiệm) thực. Chúng kiểm tra xem các lớp của bạn có được nối dây đúng cách không. Chậm hơn, nhưng chúng bắt một loại lỗi mà unit tests không thể bắt được.
• End-to-end tests: Điều khiển toàn bộ ứng dụng như một khách hàng bên ngoài. Chúng thực tế nhất nhưng cũng đắt đỏ nhất để chạy và duy trì. Thường được dành riêng cho các đường dẫn quan trọng.

Dưới đây là cách kiểm tra thực sự trông như thế nào trong thực hành, bắt đầu với các endpoint kiểm tra sức khỏe:

import pytest
from unittest.mock import AsyncMock
from fastapi.testclient import TestClient
from main import app
from app.dependencies import get_db

def test_health_endpoint():
    client = TestClient(app)
    response = client.get("/health")

    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "healthy"
    assert data["version"] == "1.0.0"

Điều này rất đơn giản — khởi chạy TestClient, truy cập endpoint, xác nhận response. TestClient là một lớp bao mỏng quanh httpx chạy ứng dụng ASGI của bạn trong quy trình. Không có máy chủ thực sự nào được khởi động, không có cổng nào được mở. Nó rất nhanh.

Phần thú vị hơn là những gì xảy ra khi bạn cần kiểm tra một endpoint phụ thuộc vào cơ sở dữ liệu:

def test_health_db_success():
    mock_db = AsyncMock()
    mock_db.check_connection.return_value = None

    app.dependency_overrides[get_db] = lambda: mock_db

    client = TestClient(app)
    response = client.get("/health/db")

    app.dependency_overrides.clear()

    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "healthy"
    assert data["dependencies"]["database"] == "connected"

Đây là nơi dependency_overrides phát huy tác dụng, và tôi đã mất một lúc để thực sự hiểu điều gì đang diễn ra ở đây. Khi FastAPI xử lý một request, nó giải quyết các dependencies bằng cách gọi các hàm đã đăng ký với Depends(). dependency_overrides cho phép bạn nói với FastAPI: "cho bài kiểm tra này, bất cứ khi nào bạn sẽ gọi get_db, hãy gọi lambda này và sử dụng mock mà nó trả về".

Bạn không đang vá lỗi cơ sở dữ liệu本身 — bạn đang vá điểm tiêm. Route handler nhận mock_db chính xác như cách nó sẽ nhận một thể hiện Database thực, và nó không thể nói lên sự khác biệt. check_connection là một AsyncMock, nên việc await nó hoạt động mà không phàn nàn, và .return_value = None có nghĩa là nó giải quyết thành công — mô phỏng một kết nối khỏe mạnh.

Trường hợp thất bại cũng quan trọng không kém để kiểm tra, và đó là nơi side_effect phát huy tác dụng:

def test_health_db_failure():
    mock_db = AsyncMock()
    mock_db.check_connection.side_effect = Exception("Database connection failed")

    app.dependency_overrides[get_db] = lambda: mock_db

    client = TestClient(app)
    response = client.get("/health/db")

    app.dependency_overrides.clear()

    assert response.status_code == 503
    data = response.json()
    assert data["detail"]["status"] == "unhealthy"
    assert "error" in data["detail"]

side_effect bảo AsyncMock ném ra ngoại lệ đó khi check_connection được await, thay vì trả về một giá trị. Điều này kiểm tra khối bắt (catch block) trong route của bạn — rằng một sự cố thực sự thực sự trả về 503 và hình dạng lỗi đúng. Nếu không có bài kiểm tra này, bạn có thể làm hỏng xử lý lỗi của mình và không biết cho đến khi một sự cố cơ sở dữ liệu thực đánh vào production.

Một điều lớn mà tôi thường xuyên bị bối rối khi mới bắt đầu: app.dependency_overrides.clear() ở cuối mỗi bài kiểm tra không phải là dọn dẹp tùy chọn. Nếu bạn quên nó, sự ghi đè từ một bài kiểm tra sẽ chảy sang bài tiếp theo, và bạn kết thúc với các bài kiểm tra vượt qua hoặc thất bại dựa trên thứ tự thực thi — một trong những lỗi khó hiểu nhất mà bạn có thể trong một bộ kiểm tra. Trong thực tế, hãy sử dụng fixture của pytest để xử lý việc này tự động:

@pytest.fixture(autouse=True)
def clear_overrides():
    yield
    app.dependency_overrides.clear()

autouse=True có nghĩa là fixture này chạy cho mọi bài kiểm tra trong phạm vi tự động, mà không cần bạn nhớ phải bao gồm nó. Mọi thứ sau yield là dọn dẹp — nó chạy sau khi bài kiểm tra hoàn tất, dù nó vượt qua hay thất bại.

Những điều tôi chưa cần (nhưng biết nên đặt ở đâu)

Một phần của việc tìm ra điều này là học cách không thêm mọi thứ trước khi chúng cần thiết. Đây là các tiện ích mở rộng tôi biết về nhưng chỉ dùng khi có tín hiệu yêu cầu:

• Background tasks: Khi một thứ như gửi email xác nhận đang chặn response HTTP. FastAPI có BackgroundTasks nhẹ nhàng cho các trường hợp đơn giản. Celery + Redis khi bạn cần thử lại, lên lịch, hoặc một quy trình worker phù hợp.
• Cache layer (Lớp lưu trữ đệm): Khi một endpoint đọc nhiều bị chậm dưới tải và dữ liệu hiếm khi thay đổi. Redis phía trước lời gọi service. Tôi chưa cần cái này ở các dự án tôi đã làm, nhưng tôi biết chỗ để lắp nó vào.
• External clients (app/clients/): Bất cứ lúc nào ứng dụng của tôi gọi một API bên thứ ba (Stripe, SendGrid, dịch vụ nội bộ), chúng có file riêng dưới clients/. Giữ những thứ đó ra khỏi services để services có thể kiểm tra được.
• Migrations (Di chuyển): Alembic, ngay từ khi có dữ liệu thực. Đây là điều tôi học theo cách khó khăn. Quản lý các thay đổi lược đồ mà không có file migration thì ổn cho đến khi nó đột nhiên không ổn nữa.

Điều tôi thực sự học được từ tất cả những điều này

Điều đã thay đổi cách tôi nghĩ về vấn đề này không phải là bất kỳ cấu trúc thư mục cụ thể nào — đó là ý tưởng rằng mỗi lớp nên có một công việc rõ ràng và các giới hạn rõ ràng về những gì nó được phép làm.

Routes chuyển ngữ HTTP. Services giữ logic. Repositories giữ các truy vấn. Settings giữ cấu hình. Khi từng mảnh ở đúng làn đường của nó, codebase vẫn dễ điều hướng và mọi thứ dễ tìm. Khi chúng không, bạn kết thúc nơi tôi bắt đầu: code hoạt động nhưng mất 20 phút để định hướng.

Nếu bạn đã cấu trúc điều này khác biệt — đặc biệt là quanh ranh giới service/repository — tôi thực sự muốn nghe cách bạn tiếp cận nó trong các bình luận.