고급 의존성

고급 의존성 (Advanced Dependencies)

지금까지 살펴본 의존성들은 모두 고정된 함수나 클래스였어요.

하지만 경우에 따라서는, 여러 개의 서로 다른 함수나 클래스를 선언하지 않고도 의존성에 파라미터를 설정할 수 있으면 좋을 때가 있어요.

쿼리 파라미터 q에 특정 고정 내용이 들어 있는지 검사하는 의존성을 하나 만들고 싶다고 상상해 보세요.

그리고 그 고정 내용을 파라미터로 바꿀 수 있게 하고 싶은 거예요.

출처: 공식문서

"호출 가능한(callable)" 인스턴스

파이썬에는 클래스의 인스턴스를 "호출 가능(callable)"하게 만드는 방법이 있어요.

클래스 자체(이미 호출 가능하죠)가 아니라, 그 클래스의 인스턴스를 말해요.

이렇게 하려면 __call__ 메서드를 선언하면 돼요.

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

이 경우, 이 __call__이 바로 FastAPI가 추가 파라미터와 하위 의존성을 검사할 때 사용하는 메서드예요. 그리고 나중에 path operation 함수 의 파라미터에 값을 전달할 때 호출되는 것도 이 메서드예요.

인스턴스 파라미터 설정하기

이제 __init__을 사용해서 인스턴스의 파라미터를 선언할 수 있어요. 이 파라미터로 의존성을 "파라미터화"할 수 있죠.

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

이 경우, FastAPI는 __init__을 건드리거나 신경 쓰지 않아요. 우리가 코드에서 직접 사용하기 때문이죠.

인스턴스 만들기

이 클래스의 인스턴스는 이렇게 만들 수 있어요.

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

이렇게 하면 의존성을 "파라미터화"할 수 있어요. 이제 그 안에 "bar"가 checker.fixed_content라는 속성으로 들어 있죠.

인스턴스를 의존성으로 사용하기

그러면 이 checker를 Depends(FixedContentQueryChecker) 대신 Depends(checker)에서 사용할 수 있어요. 의존성이 클래스 자체가 아니라 인스턴스인 checker이기 때문이에요.

의존성을 해결할 때 FastAPI는 이 checker를 이렇게 호출해요.

checker(q="somequery")

...그리고 반환된 값을 path operation 함수 의 파라미터 fixed_content_included로 전달해요.

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

팁

이 모든 게 좀 억지스러워 보일 수 있어요. 그리고 아직 유용한지 잘 안 보일 수도 있죠.

이 예시들은 의도적으로 단순하게 만든 거예요. 하지만 모든 것이 어떻게 동작하는지 보여주죠.

보안(Security)에 관한 장에는 이와 같은 방식으로 구현된 유틸리티 함수들이 있어요.

이 모든 것을 이해했다면, 이미 보안용 유틸리티 도구들이 내부에서 어떻게 동작하는지 아는 셈이에요.

`yield`와 `HTTPException`, `except`, Background Tasks를 쓰는 의존성

경고

이런 기술적 세부 사항은 아마 필요 없을 거예요.

이 세부 사항들은 주로 0.121.0보다 오래된 yield 의존성을 쓰는 FastAPI 애플리케이션이 있고 이슈를 겪고 있을 때 유용해요.

yield를 쓰는 의존성은 여러 사용 사례를 다루고 일부 문제를 고치기 위해 시간이 지나며 발전해 왔어요. 무엇이 바뀌었는지 요약하면 이래요.

`yield`와 `scope`를 쓰는 의존성

버전 0.121.0에서 FastAPI는 yield를 쓰는 의존성에 대해 Depends(scope="function") 지원을 추가했어요.

Depends(scope="function")을 사용하면 yield 이후의 종료 코드가 path operation 함수 가 끝난 직후, 응답이 클라이언트에게 보내지기 전에 실행돼요.

그리고 Depends(scope="request")(기본값)를 사용하면 yield 이후의 종료 코드가 응답이 보내진 후에 실행돼요.

이에 대한 더 자세한 내용은 yield 의존성 - 조기 종료와 scope 문서에서 읽을 수 있어요.

`yield`와 `StreamingResponse`를 쓰는 의존성, 기술적 세부 사항

FastAPI 0.118.0 이전에는, yield를 쓰는 의존성을 사용하면 종료 코드가 path operation 함수 가 반환된 직후, 그러나 응답을 보내기 바로 전에 실행됐어요.

의도는 응답이 네트워크를 통해 전달되기를 기다리는 동안 리소스를 필요 이상으로 오래 잡고 있지 않게 하려는 거였어요.

이 변경은 또 이런 뜻이기도 했어요. StreamingResponse를 돌려줬다면 yield 의존성의 종료 코드는 이미 실행됐을 거라는 거죠.

예를 들어 yield 의존성에 데이터베이스 세션이 있다면, 스트리밍하는 동안 StreamingResponse가 그 세션을 사용할 수 없게 돼요. yield 이후의 종료 코드에서 세션이 이미 닫혔을 테니까요.

이 동작은 0.118.0에서 되돌려져서, yield 이후의 종료 코드가 응답이 보내진 후에 실행되도록 바뀌었어요.

참고

아래에서 보게 되겠지만, 이 동작은 버전 0.106.0 이전의 동작과 매우 비슷해요. 다만 여러 개선 사항과 모서리(corner case) 버그 수정이 더해졌죠.

조기 종료 코드가 필요한 사용 사례

응답을 보내기 전에 yield 의존성의 종료 코드를 실행하는 옛 동작이 도움이 되는, 특정 조건을 가진 사용 사례가 몇 가지 있어요.

예를 들어, yield 의존성에서 사용자를 검증하는 데만 데이터베이스 세션을 사용하고, path operation 함수 에서는 데이터베이스 세션을 다시는 사용하지 않는 코드가 있다고 상상해 보세요. 의존성에서만 사용하고요. 그리고 응답이 보내지는 데 오래 걸려요. 데이터를 천천히 보내는데 어떤 이유로든 데이터베이스는 사용하지 않는 StreamingResponse 같은 경우죠.

이런 경우 데이터베이스 세션이 응답이 보내지는 게 끝날 때까지 잡혀 있어요. 하지만 사용하지 않는다면 잡고 있을 필요가 없죠.

이렇게 생겼을 수 있어요.

import time
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine

engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")


def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

종료 코드, 즉 Session의 자동 닫힘은 이 부분이에요.

# 위 코드는 생략됐어요 👆

def get_session():
    with Session(engine) as session:
        yield session

# 아래 코드는 생략됐어요 👇

...이 종료 코드는 응답이 느린 데이터를 다 보낸 뒤에 실행돼요.

@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

하지만 generate_stream()은 데이터베이스 세션을 사용하지 않으므로, 응답을 보내는 동안 세션을 열어 둘 필요는 사실 없어요.

SQLModel(또는 SQLAlchemy)을 쓰는 이런 특정 사용 사례가 있다면, 더 이상 필요 없을 때 세션을 명시적으로 닫으면 돼요.

def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")
    session.close()

이렇게 하면 세션이 데이터베이스 연결을 해제해서, 다른 요청들이 그 연결을 사용할 수 있어요.

yield 의존성에서 조기 종료가 필요한 다른 사용 사례가 있다면, GitHub Discussion Question을 만들어서 구체적인 사용 사례와 yield 의존성의 조기 닫힘이 어떤 도움이 되는지 알려주세요.

yield 의존성의 조기 닫힘을 위한 설득력 있는 사용 사례들이 있다면, 조기 닫힘을 선택적으로 켤 수 있는 새로운 방법을 추가하는 것을 고려해 볼게요.

`yield`와 `except`를 쓰는 의존성, 기술적 세부 사항

FastAPI 0.110.0 이전에는, yield를 쓰는 의존성을 사용하고 그 의존성 안에서 except로 예외를 잡았는데 예외를 다시 던지지 않았다면, 그 예외는 자동으로 예외 핸들러나 내부 서버 오류 핸들러로 던져지거나/전달됐어요.

이 동작은 0.110.0에서 바뀌어서, 핸들러가 없는 전달된 예외(내부 서버 오류)에서 처리되지 않은 메모리 소비를 고치고 일반 파이썬 코드의 동작과 일관되게 만들었어요.

Background Tasks와 `yield`를 쓰는 의존성, 기술적 세부 사항

FastAPI 0.106.0 이전에는 yield 이후에 예외를 던지는 게 불가능했어요. yield 의존성의 종료 코드가 응답이 보내진 후에 실행됐기 때문에, 예외 핸들러는 이미 실행됐을 테니까요.

이렇게 설계된 주된 이유는 의존성이 "산출(yield)"한 객체들을 background task 안에서 사용할 수 있게 하려는 거였어요. 종료 코드가 background task들이 끝난 후에 실행될 테니까요.

이 동작은 FastAPI 0.106.0에서, 응답이 네트워크를 통해 전달되기를 기다리는 동안 리소스를 잡고 있지 않게 하려는 의도로 바뀌었어요.

팁

게다가 background task는 보통 독립적인 로직의 집합이라서, 자체 리소스(예: 자체 데이터베이스 연결)로 별도로 처리해야 해요.

그래서 이렇게 하면 아마 더 깔끔한 코드를 만들 수 있을 거예요.

이 동작을 믿고 쓰고 있었다면, 이제 background task를 위한 리소스는 background task 안에서 만들어야 해요. 그리고 yield 의존성의 리소스에 의존하지 않는 데이터만 내부적으로 사용해야 해요.

예를 들어 같은 데이터베이스 세션을 사용하는 대신, background task 안에서 새 데이터베이스 세션을 만들고 그 새 세션을 사용해서 데이터베이스에서 객체를 얻어야 해요. 그리고 데이터베이스의 객체를 background task 함수의 파라미터로 전달하는 대신, 그 객체의 ID를 전달하고 background task 함수 안에서 다시 객체를 얻는 식으로요.

더 알아보기 (Learn more)

이 문서는 FastAPI 공식 문서 - Advanced Dependencies를 한국어로 정리한 번역이에요. 원문에서 최신 내용과 더 다양한 예시를 확인하세요.

고급 의존성