Bài 3: Dependency Injection trong FastAPI

Tham khảo trong project: admin/src/auth.py, admin/src/deps.py, admin/src/db/engine.py

---

1. Dependency Injection (Tiêm phụ thuộc) thực chất là gì?

Để thấy rõ sức mạnh của DI, hãy xem thử nếu không dùng DI (viết code kiểu ngây ngô) thì trông sẽ ra sao:

@router.get("/audio")
async def list_audio():
    # ❌ Tự khởi tạo "đồ nghề" trực tiếp bên trong hàm xử lý
    session = create_session()           # Tự mở kết nối DB
    repo = SqliteAudioRepository()       # Gắn chết với SQLite
    service = AudioService(repo)         # Tự nhét repo vào service
    return await service.list_all(session)

Cách viết này mang đến 3 nỗi đau ngay lập tức:

1. Không thể viết Test: Bạn bị dính chặt với Database thật. Không có cách nào tráo một cái DB giả (mock) vào để chạy test được.
2. Sửa một chỗ, vỡ muôn nơi: Nếu sếp yêu cầu đổi DB từ SQLite sang PostgreSQL, bạn sẽ phải lóc cóc đi tìm và sửa từng cái handler một.
3. Quản lý tài nguyên lỏng lẻo: Mở session thì dễ, nhưng lỡ logic sinh lỗi giữa chừng mà bạn quên close session thì sớm muộn DB cũng sập.

Và đây là khi dùng DI (với "phép thuật" Depends của FastAPI):

@router.get("/audio")
async def list_audio(
    session: AsyncSession = Depends(get_session),   # Framework tự động nhét vào
    svc: AudioService = Depends(get_audio_service), # Framework tự động nhét vào
):
    return await svc.list_all(session)

Giờ đây, FastAPI đóng vai trò như một người quản gia mẫn cán. Nó tự động tìm kiếm, khởi tạo, tiêm (inject) các món "đồ nghề" vào cho bạn dùng, và dùng xong nó sẽ tự động dọn dẹp.

---

2. Cách tạo một Dependency đơn giản

Hàm khởi tạo (Factory function)

Đây là cách bạn "dạy" FastAPI cách lắp ráp một món đồ (ví dụ như Service).

# admin/src/routers/audio.py
def get_audio_service() -> AudioService:
    """Hàm Factory — chuyên lo việc lắp ráp service với các repository cụ thể."""
    return AudioService(
        SqliteAudioRepository(),
        SqliteCategoryRepository(),
        SqliteEventRepository(),
    )

@router.get("")
async def list_audio(
    svc: AudioService = Depends(get_audio_service),  # FastAPI sẽ gọi hàm factory này ở mỗi request
):
    ...

Dependency có chức năng "tự dọn dẹp" (Async Generator)

# admin/src/db/engine.py
async def get_session() -> AsyncIterator[AsyncSession]:
    """Cung cấp AsyncSession cho mỗi request và tự động đóng khi xong việc."""
    async with AsyncSessionLocal() as session:
        yield session
        # ← Request xử lý xong, lệnh `async with` sẽ tự động close session một cách an toàn

# Cách dùng trong router
@router.get("/audio")
async def list_audio(
    session: AsyncSession = Depends(get_session),
    # Bạn cứ xài thoải mái, session sẽ tự động close khi handler trả về kết quả
):
    ...

Bí kíp yield: Từ khóa này biến hàm của bạn thành một quy trình 2 nhịp. FastAPI sẽ chạy đoạn code trước yield để lấy đồ đưa cho bạn dùng. Đợi bạn xử lý xong request, nó quay lại chạy nốt đoạn code sau yield để dọn dẹp rác. Về bản chất nó giống hệt câu lệnh with, chỉ khác là bạn trao quyền kiểm soát cho framework.

📌 Góc "thực chiến" — Transaction nằm ở đâu? Project này không dùng chiêu bọc nguyên một cục session.begin() ở tầng dependency. Thay vào đó, mỗi hàm của repository sẽ tự gọi await session.commit() sau khi thay đổi dữ liệu xong.

Pattern này đơn giản, dễ viết, nhưng đòi hỏi một sự đánh đổi: Nếu Service của bạn gọi 3 hành động thay đổi DB, mà hành động thứ 3 bị văng lỗi, thì 2 hành động trước đó đã bị lưu (persist) mất rồi! Lớp Service sẽ phải tự lo việc dọn dẹp "bãi chiến trường" (chúng ta sẽ đào sâu cái này ở Bài 4, mục "Transaction order là business rule").

Ví dụ cụ thể — audio_repo.py tự chốt đơn (commit) ngay trong từng hàm:

# admin/src/repositories/audio_repo.py
class SqliteAudioRepository(AudioRepository):
    async def create(
        self,
        session: AsyncSession,
        *,
        filename: str,
        title: str,
        title_normalized: str,
        size_bytes: int,
        duration_seconds: int,
        category_id: int | None,
        event_id: int | None,
        uploaded_at: str,
    ) -> dict[str, Any]:
        result = await session.execute(
            insert(audio)
            .values(
                filename=filename,
                title=title,
                title_normalized=title_normalized,
                size_bytes=size_bytes,
                duration_seconds=duration_seconds,
                category_id=category_id,
                event_id=event_id,
                uploaded_at=uploaded_at,
            )
            .returning(audio)                  # Mẹo: lấy luôn row vừa insert, khỏi mất công SELECT lại
        )
        row = result.fetchone()
        await session.commit()                 # ← Chốt đơn ngay tại repo method
        return _row_to_dict(row)

    async def delete(self, session: AsyncSession, id: int) -> bool:
        result = await session.execute(delete(audio).where(audio.c.id == id))
        await session.commit()                 # ← 1 thao tác = 1 Transaction tự commit
        return (result.rowcount or 0) > 0

Mỗi hàm là một transaction độc lập. Tầng Service cứ thế gọi repo.create() / repo.delete() mà không bận tâm chuyện commit. Trade-off của cách làm này được giải thích kỹ ở Bài 4 mục 9 và Bài 5 mục 8.

---

3. Chuỗi Dependency (Dependency Chain)

Các Dependency hoàn toàn có thể "dựa dẫm" vào nhau. FastAPI cực kỳ thông minh, nó sẽ tự động vẽ ra một cái sơ đồ và lấy đồ cho bạn theo đúng thứ tự:

# Tầng 1: Lấy DB session
async def get_session() -> AsyncSession: ...

# Tầng 2: Lấy User Repo (Cần session DB để chạy)
def get_user_repo() -> UserRepository:
    return SqliteUserRepository()

# Tầng 3: Lấy thông tin user hiện tại (Cần cả Session và User Repo)
async def get_current_user(
    cookie_token: str | None = Cookie(default=None, alias="kcds_session"),
    session: AsyncSession = Depends(get_session),       # ← Gọi Tầng 1
    user_repo: UserRepository = Depends(get_user_repo), # ← Gọi Tầng 2
) -> dict:
    email = verify_session(cookie_token)  # Giải mã JWT
    return await user_repo.get_by_email(session, email)

# Tầng 4: Kiểm duyệt quyền (Cần thông tin user)
def require_editor(
    user: dict = Depends(get_current_user),  # ← Gọi Tầng 3
) -> dict:
    if user["role"] not in ("admin", "sub_admin"):
        raise HTTPException(403, "Không đủ quyền")
    return user

# Tầng 5: Router — Đích đến cuối cùng
@router.post("/audio")
async def upload_audio(
    user: dict = Depends(require_editor),  # ← Gọi Tầng 4
    session: AsyncSession = Depends(get_session),
):
    # Vào đến đây là user đã được kiểm tra quyền kỹ càng
    ...

FastAPI sẽ tự động chạy ngầm như sau:

1. Thấy upload_audio cần require_editor → mò tìm require_editor.
2. Thấy require_editor cần get_current_user → mò tìm tiếp.
3. Thấy get_current_user cần get_session và get_user_repo → OK, tạo DB Session trước.
4. Xong xuôi, mang Session chạy ngược lên các tầng trên để nhồi vào hàm.
5. Router chạy xong → tự động quay lại dọn dẹp get_session (đóng kết nối).

Mẹo hay: Dù get_session bị réo tên ở cả handler lẫn get_current_user, FastAPI sẽ lưu tạm (cache) nó lại trong vòng đời của 1 request. Bạn yên tâm là nó chỉ tạo ĐÚNG 1 session chứ không mở 2 kết nối thừa thãi.

---

4. dependencies=[] — Khóa cửa toàn bộ Router

Thay vì cứ phải lóc cóc dán bùa Depends(require_user) vào từng cái endpoint một:

# ❌ Cách viết cồng kềnh, lặp code
@router.get("/audio", dependencies=[Depends(require_user)])
async def list_audio(...): ...

@router.post("/audio", dependencies=[Depends(require_user)])
async def upload_audio(...): ...

Hãy khóa cửa ngay từ ngoài cổng (Router level):

# ✅ Áp dụng một nhát cho cả router luôn
router = APIRouter(dependencies=[Depends(require_user)])

@router.get("")      # Tự động bắt require_user
async def list_audio(...): ...

@router.post("")     # Cũng tự động bắt require_user
async def upload(...): ...

Trường hợp cần chia quyền chi tiết cho từng API (Mixed Auth) thì mới set trên từng endpoint:

# admin/src/routers/audio.py
@router.get("", dependencies=[Depends(require_user)])           # Ai cũng xem được
async def list_audio(...): ...

@router.post("", dependencies=[Depends(require_editor)])        # Chỉ editor được đăng nhạc
async def upload_audio(...): ...

@router.delete("/{id}", dependencies=[Depends(require_editor)]) # Chỉ editor được xóa nhạc
async def delete_audio(...): ...

---

5. Dùng chung Factory thì ném vào deps.py

Khi một hàm Factory được sử dụng "ké" bởi từ 2 router trở lên, đừng copy-paste, hãy gom nó vào file deps.py:

# admin/src/deps.py
def get_schedule_service() -> ScheduleService:
    """Dùng chung cho cả router /schedule và /playlists."""
    return ScheduleService(
        SqliteScheduleRepository(),
        SqlitePlaylistRepository(),
        SqliteBotStateRepository(),
        ChannelsService(SqliteDiscordChannelsRepository()),
    )

def get_user_service() -> UserService:
    """Dùng chung cho router /auth (xử lý logout) và /users (CRUD)."""
    return UserService(get_user_repo())

Quy tắc ngầm: Factory nào chỉ phục vụ duy nhất 1 router → Cứ để yên nó trong file router đó. Khéo co thì ấm, đừng lôi hết mọi thứ ra deps.py khiến file này biến thành một bãi rác khổng lồ.

---

6. Tại sao DI lại là "cứu tinh" khi viết Test?

Hãy xem sự mầu nhiệm của DI khi bạn cần viết Unit Test:

# Code thật đang chạy
@router.get("/audio")
async def list_audio(svc=Depends(get_audio_service)):
    return await svc.list_all(session)

# Khi viết Test — Chúng ta "Tráo đồ" (Override dependency)
from unittest.mock import AsyncMock, Mock

def get_mock_service():
    # Tạo ra một service giả (mock) cực kỳ nhẹ bén
    svc = Mock(spec=AudioService)
    svc.list_all = AsyncMock(return_value=[{"id": 1, "title": "Bản nhạc Test"}])
    return svc

# Ép FastAPI dùng đồ giả thay vì đồ thật
app.dependency_overrides[get_audio_service] = get_mock_service

# Giờ thì chạy test bét nhè mà không cần bận tâm đến Database thật!
async def test_list_audio():
    response = await client.get("/api/audio")
    assert response.json() == {"data": [{"id": 1, "title": "Bản nhạc Test"}]}

---

Ánh xạ vào Nguyên lý tổng quát (Bài 0)

Mảnh ghép trong bài	Thuộc Nguyên lý nào?	Cách mang đi áp dụng nơi khác
Dùng Depends(get_session)	IoC (Đảo ngược luồng điều khiển)	Framework quyết định khi nào gọi factory. Nó giống hệt cơ chế DI của Angular, @Autowired của Spring hay @Inject của NestJS.
Service nhận Repo qua constructor	DIP (Đảo ngược phụ thuộc)	Phụ thuộc vào cái vỏ interface (AudioRepository ABC). Đây là nền tảng sống còn để viết Unit Test cực nhàn (Bài 9 mục 4).
Gom factory vào deps.py	DRY có kỷ luật	Chỉ gom khi nó thực sự được xài chung. Đừng gom vội — hãy nhớ "Luật quá tam ba bận" ở Bài 0.
Khai báo dependencies=[...] ở Router	SoC (Tách biệt mối quan tâm)	Tách bạch phần kiểm tra quyền ra khỏi phần logic xử lý chính. Hoạt động y chang Middleware trong Express hay Guards trong NestJS.

Đừng nghĩ DI là "chén thánh" — Lạm dụng là toang

Ở Bài 10 mục 6, mình có chỉ ra "căn bệnh" lạm dụng DI. Chẳng hạn, đừng rảnh rỗi viết hàm Depends(get_datetime_now) chỉ để lấy giờ hệ thống hiện tại datetime.now(). Chỉ nên dùng DI cho: Các dịch vụ bên ngoài (HTTP Client, Database), các lớp logic lõi (Repository, Service), hoặc các biến cấu hình (Config). Không nên dùng DI cho các hàm tiện ích nhỏ nhặt thuần túy của Python.

---

Bài tập thực hành

Hãy thử tự tay xây dựng một chuỗi Dependency cho API quản lý bài viết (Blog):

# 1. Dependency cấp DB
async def get_db() -> AsyncIterator[Session]:
    async with SessionLocal() as session:
        yield session

# 2. Dependency cấp Auth
async def get_current_user(
    token: str | None = Header(default=None, alias="Authorization"),
    db: Session = Depends(get_db),
) -> dict:
    if not token:
        raise HTTPException(401, "Bạn chưa đăng nhập")
    # Tưởng tượng có logic giải mã token ở đây...
    return user

# 3. Dependency cấp Phân Quyền
def require_author(user: dict = Depends(get_current_user)) -> dict:
    if user["role"] != "author":
        raise HTTPException(403, "Chỉ tác giả mới được đăng bài")
    return user

# 4. Gắn chuỗi này vào Router
@router.post("/posts", dependencies=[Depends(require_author)])
async def create_post(body: PostCreate, db=Depends(get_db)):
    ...

Thử thách tư duy:

1. Điều gì sẽ xảy ra với get_session sau khi handler trả về kết quả cho user? Tại sao lại thế?
2. Nếu get_db được gọi tới 2 lần (ở cả handler lẫn get_current_user), FastAPI sẽ mở mấy kết nối Database?
3. Tại sao khai báo dependencies=[Depends(...)] trên đầu router lại nhìn "sạch sẽ" hơn là nhét nó vào tham số của từng hàm?