FastAPI Dependency Injection 패턴

FastAPI는 강력한 의존성 주입 시스템을 제공하여 코드의 재사용성과 테스트 용이성을 높여줍니다. 이 글에서는 다양한 의존성 주입 패턴을 알아봅니다.

Environment

$ python --version
Python 3.11.5

$ pip show fastapi
Name: fastapi
Version: 0.104.1

$ pip show uvicorn
Name: uvicorn
Version: 0.24.0

Problem: 코드 중복과 테스트 어려움

# Without dependency injection - lots of duplication
from fastapi import FastAPI
import databases
import sqlalchemy

app = FastAPI()

DATABASE_URL = "sqlite:///./test.db"
database = databases.Database(DATABASE_URL)
engine = sqlalchemy.create_engine(DATABASE_URL)

@app.get("/users/")
async def get_users():
    # Database connection code repeated everywhere
    query = users.select()
    return await database.fetch_all(query)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Same database connection code
    query = users.select().where(users.c.id == user_id)
    return await database.fetch_one(query)

Analysis: Dependency Injection 동작 원리

from fastapi import Depends, FastAPI

app = FastAPI()

# Dependency function
def common_dependency():
    return {"data": "shared"}

# Dependency injected into route
@app.get("/")
async def route(dependency: dict = Depends(common_dependency)):
    return dependency

# Dependencies can be chained
def dep_a():
    return "A"

def dep_b(a: str = Depends(dep_a)):
    return f"{a} -> B"

@app.get("/chain/")
async def chain(b: str = Depends(dep_b)):
    return b  # Returns "A -> B"

Solution: Dependency Injection 패턴

기본 의존성 정의

from fastapi import Depends, FastAPI, HTTPException
from typing import Optional
import databases

app = FastAPI()

# Database dependency
DATABASE_URL = "sqlite:///./test.db"
database = databases.Database(DATABASE_URL)

async def get_db():
    """Database session dependency"""
    try:
        yield database
    finally:
        # Cleanup if needed
        pass

# User dependency
async def get_current_user(db = Depends(get_db)):
    """Get current authenticated user"""
    # In real app, decode JWT token here
    user = await db.fetch_one("SELECT * FROM users WHERE id = 1")
    if not user:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return user

@app.get("/users/me/")
async def read_users_me(current_user = Depends(get_current_user)):
    return current_user

Class-based Dependencies

from fastapi import Depends, FastAPI
from typing import Optional

class PaginationParams:
    def __init__(self, skip: int = 0, limit: int = 100):
        self.skip = skip
        self.limit = min(limit, 100)  # Max 100 items

class QueryFilter:
    def __init__(
        self,
        search: Optional[str] = None,
        category: Optional[str] = None,
        sort_by: str = "created_at"
    ):
        self.search = search
        self.category = category
        self.sort_by = sort_by

app = FastAPI()

@app.get("/items/")
async def list_items(
    pagination: PaginationParams = Depends(),
    filters: QueryFilter = Depends()
):
    return {
        "skip": pagination.skip,
        "limit": pagination.limit,
        "filters": {
            "search": filters.search,
            "category": filters.category,
            "sort_by": filters.sort_by
        }
    }

의존성 오버라이드 (테스트용)

from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient

app = FastAPI()

def get_settings():
    return {"debug": False, "database_url": "production.db"}

@app.get("/config/")
async def get_config(settings = Depends(get_settings)):
    return settings

# For testing - override dependency
def override_get_settings():
    return {"debug": True, "database_url": "test.db"}

app.dependency_overrides[get_settings] = override_get_settings

client = TestClient(app)
response = client.get("/config/")
print(response.json())  # {"debug": true, "database_url": "test.db"}

의존성 스코프

from fastapi import FastAPI, Depends
from functools import lru_cache

app = FastAPI()

# Cached dependency - called once per request
@lru_cache()
def expensive_dependency():
    print("Computing expensive value...")
    return {"value": 42}

@app.get("/cached/")
async def cached_route(dep = Depends(expensive_dependency)):
    return dep

# Sub-dependencies
def dep_a():
    return "A"

def dep_b():
    return "B"

def combined(a: str = Depends(dep_a), b: str = Depends(dep_b)):
    return f"{a}-{b}"

@app.get("/combined/")
async def combined_route(c: str = Depends(combined)):
    return c

Yield Dependencies (리소스 관리)

from fastapi import FastAPI, Depends
from typing import Generator

app = FastAPI()

def get_db_session() -> Generator:
    """Database session with automatic cleanup"""
    session = Session()
    try:
        yield session
    finally:
        session.close()

def get_cache() -> Generator:
    """Cache connection with cleanup"""
    cache = Redis()
    try:
        yield cache
    finally:
        cache.disconnect()

@app.get("/data/")
async def get_data(
    db = Depends(get_db_session),
    cache = Depends(get_cache)
):
    # Use db and cache
    cached = cache.get("data")
    if not cached:
        cached = db.query("SELECT * FROM data")
        cache.set("data", cached)
    return cached

의존성 조件부 사용

from fastapi import FastAPI, Depends, Query
from typing import Optional

app = FastAPI()

def validate_sort_order(sort: str = Query("asc", regex="^(asc|desc)$")):
    return sort

def validate_pagination(
    page: int = Query(1, ge=1),
    per_page: int = Query(10, ge=1, le=100)
):
    return {"page": page, "per_page": per_page}

@app.get("/items/")
async def list_items(
    sort: str = Depends(validate_sort_order),
    pagination: dict = Depends(validate_pagination)
):
    return {
        "sort": sort,
        "page": pagination["page"],
        "per_page": pagination["per_page"]
    }

Lessons Learned

1. 함수 의존성: 간단한 의존성은 함수로, 복잡한 의존성은 클래스로 만드는 것이 효과적입니다.

2. yield 의존성: 리소스 연결/해제가 필요하면 yield를 사용하는 의존성을 사용하세요.

3. 테스트용 오버라이드: app.dependency_overrides를 사용하여 테스트 환경에서 의존성을 쉽게 교체할 수 있습니다.

4. 의존성 체이닝: 의존성은 다른 의존성을 참조할 수 있어, 복잡한 초기화 로직을 분리할 수 있습니다.

5. 성능 고려: 자주 사용되는 의존성은 @lru_cache로 캐싱하여 중복 계산을 방지하세요.

This blog does not accept any external sponsorships, affiliate marketing, or ad revenue.