Sổ tay kỹ thuật

Cấu trúc project FastAPI – router, settings, DI, testing

concept#python#fastapi#api· cập nhật 2026-11-06

Dùng khi API FastAPI vượt quá một file main.py và cần chia router, quản lý cấu hình bằng pydantic-settings, dependency injection và viết test.

Sửa trang này
Mục lục

Khi app còn dưới khoảng 200 dòng, một file main.py là đủ (xem FastAPI cơ bản). Lớn hơn thì chia theo router (mỗi nhóm endpoint một file), tách config, dependency và logic nghiệp vụ ra khỏi handler để test được.

Cấu trúc thư mục gợi ý

netinv/
├── pyproject.toml
├── uv.lock
├── .env.example            # mẫu biến môi trường, không commit .env thật
├── app/
│   ├── __init__.py
│   ├── main.py             # tạo FastAPI(), include_router, middleware, lifespan
│   ├── config.py           # Settings (pydantic-settings) + get_settings()
│   ├── deps.py             # các Depends dùng chung: settings, db session, auth
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── devices.py      # APIRouter(prefix="/devices")
│   │   └── health.py
│   ├── schemas/            # Pydantic model cho request/response
│   │   └── device.py
│   └── services/           # logic nghiệp vụ, gọi NetBox/SSH/DB; không import FastAPI
│       └── inventory.py
└── tests/
    └── test_devices.py

Luồng request qua các lớp

flowchart LR
  R[Request] --> M["main.py (middleware)"]
  M --> RT["routers/devices.py"]
  RT --> DP["deps.py (Depends)"]
  DP --> CF["config.py (Settings)"]
  RT --> SV["services/inventory.py"]
  SV --> EX["NetBox / DB / thiết bị"]
  RT --> SC["schemas (response_model)"]

Thành phần

File Vai trò Ghi chú
main.py Lắp ráp app Không chứa logic, chỉ include_router
routers/*.py Khai báo endpoint Mỏng: validate → gọi service → trả schema
schemas/*.py Pydantic model Tách DeviceIn / DeviceOut để không lộ field nội bộ
services/*.py Logic nghiệp vụ Hàm Python thường, test được không cần HTTP
config.py Đọc env / .env Một nguồn cấu hình duy nhất
deps.py Dependency dùng chung Dễ thay bằng dependency_overrides khi test

Settings bằng pydantic-settings

app/config.py
from functools import lru_cache
 
from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="APP_", env_file=".env", extra="ignore")
 
    env: str = "dev"                          # APP_ENV
    database_url: str = "sqlite:///./dev.db"  # APP_DATABASE_URL
    netbox_url: str = "https://netbox.example.com"
    netbox_token: SecretStr = SecretStr("")   # APP_NETBOX_TOKEN, không in ra log
 
 
@lru_cache
def get_settings() -> Settings:
    return Settings()

SecretStr hiển thị ********** khi print; lấy giá trị thật bằng .get_secret_value().

Dependency và router

app/deps.py
from typing import Annotated
 
from fastapi import Depends
 
from app.config import Settings, get_settings
 
SettingsDep = Annotated[Settings, Depends(get_settings)]
app/routers/devices.py
from fastapi import APIRouter
 
from app.deps import SettingsDep
 
router = APIRouter(prefix="/devices", tags=["devices"])
 
 
@router.get("")
async def list_devices(settings: SettingsDep) -> dict:
    return {"env": settings.env, "items": []}
app/main.py
from fastapi import FastAPI
 
from app.routers import devices
 
app = FastAPI(title="NetInventory API")
app.include_router(devices.router, prefix="/api/v1")   # -> GET /api/v1/devices

Chạy: uv run fastapi dev app/main.py.

Test với TestClient và dependency_overrides

tests/test_devices.py
from fastapi.testclient import TestClient
 
from app.config import Settings, get_settings
from app.main import app
 
 
def test_list_devices():
    app.dependency_overrides[get_settings] = lambda: Settings(env="test")
    client = TestClient(app)
    r = client.get("/api/v1/devices")
    assert r.status_code == 200
    assert r.json()["env"] == "test"
    app.dependency_overrides.clear()
uv add --dev pytest
uv run pytest -q

TestClient cần httpx (có sẵn trong fastapi[standard]). Muốn lifespan chạy trong test thì dùng with TestClient(app) as client:.

async def hay def?

Handler FastAPI chạy ở đâu Dùng khi
async def Event loop chính Gọi thư viện async: httpx.AsyncClient, asyncpg, scrapli async
def Threadpool Gọi thư viện blocking: requests, netmiko, ncclient, SQLAlchemy sync

Gọi hàm blocking (time.sleep, requests.get, SSH bằng netmiko) bên trong async def sẽ chặn mọi request khác. Không chắc thì viết def.

Khi nào dùng

  • Một file main.py: prototype, webhook nhỏ, công cụ nội bộ vài endpoint.
  • Cấu trúc trên: từ khoảng 3 nhóm endpoint trở lên, nhiều người cùng sửa, cần test và nhiều môi trường (dev/staging/prod).
  • Thêm lớp repositories/, Alembic migration: khi có DB thật với nhiều bảng.
  • Không cần tách sẵn services/ nếu handler chỉ đọc/ghi đơn giản; tách khi logic bắt đầu bị lặp hoặc cần test riêng.

Tham khảo