본문으로 건너뛰기
Advertisement

의존성 주입 (Dependency Injection)

FastAPI의 Depends는 공통 로직(인증, DB 연결, 페이지네이션)을 의존성 함수로 분리하고, 테스트 시 쉽게 교체할 수 있게 합니다.

기본 Depends​

from fastapi import FastAPI, Depends, HTTPException
from typing import Annotated

app = FastAPI()


# 의존성 함수: 공통 쿼리 파라미터 처리
def common_parameters(
    skip: int = 0,
    limit: int = 10,
    q: str | None = None,
):
    return {"skip": skip, "limit": limit, "q": q}


# 타입 힌트로 간결하게
CommonParams = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
def list_items(commons: CommonParams):
    return {"params": commons, "items": []}


@app.get("/users/")
def list_users(commons: CommonParams):
    return {"params": commons, "users": []}


# 클래스 기반 의존성
class Pagination:
    def __init__(self, page: int = 1, size: int = 20):
        self.skip = (page - 1) * size
        self.limit = size
        self.page = page

    def __repr__(self):
        return f"Pagination(page={self.page}, skip={self.skip}, limit={self.limit})"


@app.get("/products/")
def list_products(pagination: Annotated[Pagination, Depends()]):
    return {
        "page": pagination.page,
        "skip": pagination.skip,
        "limit": pagination.limit,
    }

중첩 의존성​

from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Annotated

app = FastAPI()

# 토큰 DB (시뮬레이션)
fake_tokens = {"alice-token": "Alice", "bob-token": "Bob"}
fake_users = {"Alice": {"id": 1, "role": "admin"}, "Bob": {"id": 2, "role": "user"}}


# 1단계: 토큰 추출
def get_token(x_token: Annotated[str, Header()]) -> str:
    if x_token not in fake_tokens:
        raise HTTPException(status_code=401, detail="Invalid token")
    return x_token


# 2단계: 토큰으로 사용자 조회 (1단계에 의존)
def get_current_user(token: Annotated[str, Depends(get_token)]) -> dict:
    username = fake_tokens[token]
    return fake_users[username]


# 3단계: 관리자 확인 (2단계에 의존)
def require_admin(user: Annotated[dict, Depends(get_current_user)]) -> dict:
    if user["role"] != "admin":
        raise HTTPException(status_code=403, detail="Admin only")
    return user


CurrentUser = Annotated[dict, Depends(get_current_user)]
AdminUser = Annotated[dict, Depends(require_admin)]


@app.get("/me")
def read_me(user: CurrentUser):
    return user


@app.get("/admin/dashboard")
def admin_dashboard(admin: AdminUser):
    return {"message": "Welcome Admin", "user": admin}

리소스 관리 — yield 의존성​

from fastapi import FastAPI, Depends
from typing import Annotated, Generator

app = FastAPI()


# DB 연결 시뮬레이션
class DBSession:
    def __init__(self, db_url: str):
        self.url = db_url
        self.queries: list[str] = []
        print(f"[DB] 연결: {db_url}")

    def execute(self, query: str) -> list:
        self.queries.append(query)
        return [{"id": 1, "data": query}]

    def close(self):
        print(f"[DB] 연결 해제 (쿼리 {len(self.queries)}개 실행)")


DATABASE_URL = "postgresql://localhost/mydb"


# yield 의존성: 요청 전 setup, 응답 후 teardown
def get_db() -> Generator[DBSession, None, None]:
    db = DBSession(DATABASE_URL)
    try:
        yield db
    finally:
        db.close()  # 항상 실행 (예외 발생해도)


DB = Annotated[DBSession, Depends(get_db)]


@app.get("/users/")
def list_users(db: DB):
    return db.execute("SELECT * FROM users")


@app.get("/users/{user_id}")
def get_user(user_id: int, db: DB):
    return db.execute(f"SELECT * FROM users WHERE id={user_id}")

전역 의존성​

from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Annotated

app = FastAPI()


# API 키 검증
async def verify_api_key(x_api_key: Annotated[str, Header()]):
    if x_api_key != "secret-key":
        raise HTTPException(status_code=403, detail="Invalid API Key")


# 라우터 전체에 의존성 적용
from fastapi import APIRouter

router = APIRouter(
    prefix="/api",
    dependencies=[Depends(verify_api_key)],  # 모든 라우트에 적용
)


@router.get("/data")
def get_data():
    return {"data": "protected"}


@router.get("/more-data")
def get_more_data():
    return {"data": "also protected"}


# 앱 전역 의존성
app_with_global = FastAPI(
    dependencies=[Depends(verify_api_key)]  # 모든 엔드포인트에 적용
)

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

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

app = FastAPI()


# 실제 의존성
def get_db():
    return {"type": "production_db"}


DB = Annotated[dict, Depends(get_db)]


@app.get("/items/")
def list_items(db: DB):
    return {"db_type": db["type"]}


# 테스트에서 의존성 교체
def get_test_db():
    return {"type": "test_db"}


def test_list_items():
    app.dependency_overrides[get_db] = get_test_db  # 교체

    client = TestClient(app)
    response = client.get("/items/")

    assert response.status_code == 200
    assert response.json()["db_type"] == "test_db"

    app.dependency_overrides.clear()  # 복원

정리​

개념설명
Depends(fn)의존성 함수 주입
중첩 의존성의존성이 다른 의존성을 사용
yield 의존성setup/teardown 패턴
전역 의존성router(dependencies=[...])
dependency_overrides테스트 시 의존성 교체
Annotated타입 힌트에 의존성 포함

의존성 주입은 인증, DB 세션, 공통 파라미터를 재사용하고 테스트를 용이하게 합니다.

Advertisement
Advertisement