[FastAPI] Dependency Injection

FastAPI DI, FastAPI Dependency Injection, FastAPI Depends, FastAPI 의존성 주입, sub-dependencies, dependency_overrides, yield dependency

정의

FastAPI Dependency Injection 은 Depends() 로 함수를 endpoint 에 자동 주입하는 시스템입니다. DB 세션, 인증 사용자, 설정, 서비스 계층 등을 endpoint 의 파라미터로 선언하면 FastAPI 가 알아서 해결/캐싱/lifecycle 관리를 수행합니다.

Java Spring / Angular 의 DI 컨테이너보다 훨씬 가벼우며, 함수 시그니처 = 스펙 원칙에 부합합니다.

기본 사용

from fastapi import Depends, FastAPI

app = FastAPI()

def common_query(q: str | None = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
def read_items(commons: dict = Depends(common_query)):
    return commons

common_query 는 요청마다 호출되고 그 반환값이 commons 로 주입됩니다.

Type hint 지원

Python 3.9+ Annotated 로 더 명확한 스타일:

from typing import Annotated

CommonsDep = Annotated[dict, Depends(common_query)]

@app.get("/items/")
def read_items(commons: CommonsDep):
    return commons

Class-based Dependency

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 10):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
def read_items(commons: CommonQueryParams = Depends()):
    return {"skip": commons.skip, "limit": commons.limit}

Depends() 인자 생략 시 클래스 자체를 factory 로 사용.

Sub-dependency

Dependency 가 또 다른 dependency 를 가질 수 있음. 트리 구조.

def query_extractor(q: str | None = None):
    return q

def query_or_cookie_extractor(
    q: Annotated[str | None, Depends(query_extractor)],
    last_query: Annotated[str | None, Cookie()] = None,
):
    if not q:
        return last_query
    return q

@app.get("/items/")
def read_items(query: Annotated[str, Depends(query_or_cookie_extractor)]):
    return {"q_or_cookie": query}

같은 dependency 가 한 요청에서 여러 번 참조되어도 캐시 되어 한 번만 호출.

Cache 비활성화

Depends(some_dep, use_cache=False)

Yield-based (Cleanup)

with 를 대신하는 lifecycle. DB 세션, 파일 핸들 등에 유용.

from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{id}")
def read_user(id: int, db: Annotated[Session, Depends(get_db)]):
    return db.query(User).get(id)

try/finally 는 endpoint 정상/예외 종료 모두에서 실행. 예외는 위로 전파.

Async yield

async def get_async_db():
    async with AsyncSession(engine) as session:
        yield session

Router / App level dependency

특정 endpoint 가 아니라 라우터 전체 / 앱 전체에 걸어 인증/로깅 등 공통 처리.

from fastapi import Depends

def verify_token(x_token: Annotated[str, Header()]) -> None:
    if x_token != "expected":
        raise HTTPException(status_code=401)

router = APIRouter(
    prefix="/admin",
    dependencies=[Depends(verify_token)],   # 라우터 전체
)

app = FastAPI(dependencies=[Depends(logging_middleware)])  # 앱 전체

이 경우 dependency 의 반환값은 endpoint 에 주입되지 않고 side effect (검증, 로깅) 만 수행.

인증 (OAuth2 / JWT 예시)

from fastapi.security import OAuth2PasswordBearer
from jose import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload["sub"]
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
    user = db.query(User).get(user_id)
    if not user:
        raise HTTPException(status_code=401)
    return user

def get_current_admin(user: Annotated[User, Depends(get_current_user)]) -> User:
    if not user.is_admin:
        raise HTTPException(status_code=403)
    return user

@app.get("/admin/dashboard")
def dashboard(admin: Annotated[User, Depends(get_current_admin)]):
    return {"welcome": admin.email}

OAuth2PasswordBearer 는 자동으로 Swagger UI 에 “Authorize” 버튼 추가.

dependency_overrides (테스트)

Endpoint 코드를 안 바꾸고 테스트에서 dependency 를 교체.

from fastapi.testclient import TestClient

def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)

response = client.get("/users/1")

# 테스트 후 clear
app.dependency_overrides = {}

핵심 강점: 프로덕션 코드는 순수 함수 시그니처, 테스트 인프라는 별도 파일.

Global State Anti-pattern

FastAPI DI 는 global state 를 지양 하는 방향. 다음은 안 좋은 예:

# BAD: global 로 db 를 직접 참조
db_session = SessionLocal()

@app.get("/users/{id}")
def read_user(id: int):
    return db_session.query(User).get(id)   # thread-unsafe, 테스트 불가

DI 사용:

# GOOD
@app.get("/users/{id}")
def read_user(id: int, db: Annotated[Session, Depends(get_db)]):
    return db.query(User).get(id)

서비스 계층 DI

class UserService:
    def __init__(self, db: Session, cache: Redis):
        self.db = db
        self.cache = cache

    def get(self, user_id: int) -> User:
        cached = self.cache.get(f"user:{user_id}")
        if cached:
            return User.model_validate_json(cached)
        user = self.db.query(User).get(user_id)
        self.cache.set(f"user:{user_id}", user.model_dump_json(), ex=300)
        return user

def get_user_service(
    db: Annotated[Session, Depends(get_db)],
    cache: Annotated[Redis, Depends(get_redis)],
) -> UserService:
    return UserService(db, cache)

@app.get("/users/{id}")
def read_user(
    id: int,
    users: Annotated[UserService, Depends(get_user_service)],
):
    return users.get(id)

Java Spring 스타일 3-layer (Controller / Service / Repository) 를 FastAPI 스타일로 옮긴 예.

Depends() 실행 순서

한 요청에서:

1. Path parameter, query parameter, header 파싱
2. Dependency 트리 topological sort
3. Root dependency 부터 실행, 캐시 활용
4. Endpoint 함수 실행
5. Yield-based dependency 의 cleanup (역순)

Background Task 와 DI

from fastapi import BackgroundTasks

def send_notification(email: str, message: str, service: EmailService):
    service.send(email, message)

@app.post("/send")
def send(
    background_tasks: BackgroundTasks,
    email: str,
    service: Annotated[EmailService, Depends(get_email_service)],
):
    background_tasks.add_task(send_notification, email, "hi", service)
    return {"ok": True}

주의: BackgroundTasks 는 응답 반환 후 실행. DB 세션 dependency 는 응답 시점에 이미 닫힘 -> background 에서 사용 시 새 세션 필요.

함정

관련 위키

• FastAPI - 상위 개요
• FastAPI Routing - APIRouter 의 dependencies
• FastAPI + Pydantic - Settings DI
• FastAPI Testing - dependency_overrides
• FastAPI Middleware - DI 대체 지점
• Decorator - DI 데코레이터 근간