콘텐츠로 이동

yield를 사용하는 의존성

FastAPI는 작업 완료 후 추가 단계를 수행하는 의존성을 지원합니다.

이를 구현하려면 return 대신 yield를 사용하고, 추가로 실행할 단계 (코드)를 그 뒤에 작성하세요.

각 의존성마다 yield는 한 번만 사용해야 합니다.

기술 세부사항

다음과 함께 사용할 수 있는 모든 함수:

FastAPI의 의존성으로 사용할 수 있습니다.

사실, FastAPI는 내부적으로 이 두 데코레이터를 사용합니다.

yield를 사용하는 데이터베이스 의존성

예를 들어, 이 기능을 사용하면 데이터베이스 세션을 생성하고 작업이 끝난 후에 세션을 종료할 수 있습니다.

응답을 생성하기 전에는 yield문을 포함하여 그 이전의 코드만이 실행됩니다:

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

yield된 값은 경로 처리 및 다른 의존성들에 주입되는 값 입니다:

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

yield문 다음의 코드는 응답을 생성한 후 실행됩니다:

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

async 함수와 일반 함수 모두 사용할 수 있습니다.

FastAPI는 일반 의존성과 마찬가지로 각각의 함수를 올바르게 처리할 것입니다.

yieldtry를 사용하는 의존성

yield를 사용하는 의존성에서 try 블록을 사용한다면, 의존성을 사용하는 도중 발생한 모든 예외를 받을 수 있습니다.

예를 들어, 다른 의존성이나 경로 처리의 중간에 데이터베이스 트랜잭션 "롤백"이 발생하거나 다른 오류가 발생한다면, 해당 예외를 의존성에서 받을 수 있습니다.

따라서, 의존성 내에서 except SomeException을 사용하여 특정 예외를 처리할 수 있습니다.

마찬가지로, finally를 사용하여 예외 발생 여부와 관계 없이 종료 단계까 실행되도록 할 수 있습니다.

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

yield를 사용하는 하위 의존성

모든 크기와 형태의 하위 의존성과 하위 의존성의 "트리"도 가질 수 있으며, 이들 모두가 yield를 사용할 수 있습니다.

FastAPIyield를 사용하는 각 의존성의 "종료 코드"가 올바른 순서로 실행되도록 보장합니다.

예를 들어, dependency_cdependency_b에 의존할 수 있고, dependency_bdependency_a에 의존할 수 있습니다.

from typing import Annotated

from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

이들 모두는 yield를 사용할 수 있습니다.

이 경우 dependency_c는 종료 코드를 실행하기 위해, dependency_b의 값 (여기서는 dep_b로 명명)이 여전히 사용 가능해야 합니다.

그리고, dependency_b는 종료 코드를 위해 dependency_a의 값 (여기서는 dep_a로 명명) 이 사용 가능해야 합니다.

from typing import Annotated

from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

같은 방식으로, yield를 사용하는 의존성과 return을 사용하는 의존성을 함께 사용할 수 있으며, 이들 중 일부가 다른 것들에 의존할 수 있습니다.

그리고 yield를 사용하는 다른 여러 의존성을 필요로 하는 단일 의존성을 가질 수도 있습니다.

원하는 의존성을 원하는 대로 조합할 수 있습니다.

FastAPI는 모든 것이 올바른 순서로 실행되도록 보장합니다.

기술 세부사항

파이썬의 Context Managers 덕분에 이 기능이 작동합니다.

FastAPI는 이를 내부적으로 사용하여 이를 달성합니다.

yieldHTTPException를 사용하는 의존성

yield를 사용하는 의존성에서 try 블록을 사용해 코드를 실행하고, 그 다음 finally 뒤에 종료 코드를 실행할 수 있다는 것을 보았습니다.

또한 except를 사용해 발생한 예외를 잡고 그에 대해 무언가를 할 수도 있습니다.

예를 들어, HTTPException 같은 다른 예외를 발생시킬 수 있습니다.

이는 다소 고급 기술이며, 대부분의 경우 실제로는 필요하지 않을 것입니다. 예를 들어, 경로 처리 함수 등 나머지 애플리케이션 코드 내부에서 예외 (HTTPException 포함)를 발생시킬 수 있기 때문입니다.

하지만 필요한 경우 사용할 수 있습니다. 🤓

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


data = {
    "plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"},
    "portal-gun": {"description": "Gun to create portals", "owner": "Rick"},
}


class OwnerError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except OwnerError as e:
        raise HTTPException(status_code=400, detail=f"Owner error: {e}")


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id not in data:
        raise HTTPException(status_code=404, detail="Item not found")
    item = data[item_id]
    if item["owner"] != username:
        raise OwnerError(username)
    return item
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


data = {
    "plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"},
    "portal-gun": {"description": "Gun to create portals", "owner": "Rick"},
}


class OwnerError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except OwnerError as e:
        raise HTTPException(status_code=400, detail=f"Owner error: {e}")


@app.get("/items/{item_id}")
def get_item(item_id: str, username: str = Depends(get_username)):
    if item_id not in data:
        raise HTTPException(status_code=404, detail="Item not found")
    item = data[item_id]
    if item["owner"] != username:
        raise OwnerError(username)
    return item

예외를 잡고 그에 기반해 사용자 정의 응답을 생성하려면, 사용자 정의 예외 처리기를 생성하세요.

yieldexcept를 사용하는 의존성

yield를 사용하는 의존성에서 except를 사용하여 예외를 포착하고 예외를 다시 발생시키지 않거나 (또는 새 예외를 발생시키지 않으면), FastAPI는 일반적인 Python에서와 마찬가지로 예외가 있었다는 것을 알아차릴 수 없습니다:

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("Oops, we didn't raise again, Britney 😱")


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("Oops, we didn't raise again, Britney 😱")


@app.get("/items/{item_id}")
def get_item(item_id: str, username: str = Depends(get_username)):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

이 경우, HTTPException이나 유사한 예외를 발생시키지 않기 때문에 클라이언트는 마땅히 HTTP 500 Internal Server Error 응답을 보게 되지만, 서버에는 어떤 오류였는지에 대한 로그나 다른 표시가 전혀 남지 않게 됩니다. 😱

yieldexcept를 사용하는 의존성에서 항상 raise 하기

yield가 있는 의존성에서 예외를 잡았을 때, 다른 HTTPException이나 유사한 예외를 발생시키는 것이 아니라면, 원래 예외를 다시 발생시켜야 합니다.

raise를 사용하여 동일한 예외를 다시 발생시킬 수 있습니다:

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("We don't swallow the internal error here, we raise again 😎")
        raise


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("We don't swallow the internal error here, we raise again 😎")
        raise


@app.get("/items/{item_id}")
def get_item(item_id: str, username: str = Depends(get_username)):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

이제 클라이언트는 동일한 HTTP 500 Internal Server Error 응답을 받게 되지만, 서버 로그에는 사용자 정의 InternalError가 기록됩니다. 😎

yield를 사용하는 의존성의 실행 순서

실행 순서는 아래 다이어그램과 거의 비슷합니다. 시간은 위에서 아래로 흐릅니다. 그리고 각 열은 상호 작용하거나 코드를 실행하는 부분 중 하나입니다.

sequenceDiagram

participant client as Client
participant handler as Exception handler
participant dep as Dep with yield
participant operation as Path Operation
participant tasks as Background tasks

    Note over client,operation: Can raise exceptions, including HTTPException
    client ->> dep: Start request
    Note over dep: Run code up to yield
    opt raise Exception
        dep -->> handler: Raise Exception
        handler -->> client: HTTP error response
    end
    dep ->> operation: Run dependency, e.g. DB session
    opt raise
        operation -->> dep: Raise Exception (e.g. HTTPException)
        opt handle
            dep -->> dep: Can catch exception, raise a new HTTPException, raise other exception
        end
        handler -->> client: HTTP error response
    end

    operation ->> client: Return response to client
    Note over client,operation: Response is already sent, can't change it anymore
    opt Tasks
        operation -->> tasks: Send background tasks
    end
    opt Raise other exception
        tasks -->> tasks: Handle exceptions in the background task code
    end

정보

클라이언트에는 하나의 응답만 전송됩니다. 이는 오류 응답 중 하나일 수도 있고, 경로 처리에서 생성된 응답일 수도 있습니다.

이러한 응답 중 하나가 전송된 후에는 다른 응답을 보낼 수 없습니다.

경로 처리 함수의 코드에서 어떤 예외를 발생시키면 HTTPException을 포함해 yield를 사용하는 의존성으로 전달됩니다. 대부분의 경우 해당 예외(또는 새 예외)를 yield를 사용하는 의존성에서 다시 발생시켜, 제대로 처리되도록 해야 합니다.

조기 종료와 scope

일반적으로 yield를 사용하는 의존성의 종료 코드는 클라이언트로 응답이 전송된 후에 실행됩니다.

하지만 경로 처리 함수에서 반환한 뒤에는 더 이상 해당 의존성이 필요 없다는 것을 알고 있다면, Depends(scope="function")을 사용하여 FastAPI에 경로 처리 함수가 반환된 후, 하지만 응답이 전송되기 전에 의존성을 종료(닫기)해야 한다고 알려줄 수 있습니다.

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


def get_username():
    try:
        yield "Rick"
    finally:
        print("Cleanup up before response is sent")


@app.get("/users/me")
def get_user_me(username: Annotated[str, Depends(get_username, scope="function")]):
    return username
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


def get_username():
    try:
        yield "Rick"
    finally:
        print("Cleanup up before response is sent")


@app.get("/users/me")
def get_user_me(username: str = Depends(get_username, scope="function")):
    return username

Depends()는 다음이 될 수 있는 scope 매개변수를 받습니다:

  • "function": 요청을 처리하는 경로 처리 함수 전에 의존성을 시작하고, 경로 처리 함수가 끝난 후, 하지만 응답이 클라이언트로 전송되기 전에 의존성을 종료합니다. 즉, 의존성 함수는 경로 처리 함수둘러싸며 실행됩니다.
  • "request": 요청을 처리하는 경로 처리 함수 전에 의존성을 시작하고("function"을 사용할 때와 유사), 응답이 클라이언트로 전송된 후에 종료합니다. 즉, 의존성 함수는 요청과 응답 사이클을 둘러싸며 실행됩니다.

지정하지 않고 의존성이 yield를 사용한다면, 기본 scope"request"입니다.

하위 의존성을 위한 scope

scope="request"(기본값)로 의존성을 선언하면, 모든 하위 의존성도 scope"request"여야 합니다.

하지만 scope"function"인 의존성은 scope"function"인 의존성과 "request"인 의존성을 모두 의존성으로 가질 수 있습니다.

이는 어떤 의존성이든, 종료 코드에서 하위 의존성을 계속 사용해야 할 수도 있으므로, 하위 의존성보다 먼저 종료 코드를 실행할 수 있어야 하기 때문입니다.

sequenceDiagram

participant client as Client
participant dep_req as Dep scope="request"
participant dep_func as Dep scope="function"
participant operation as Path Operation

    client ->> dep_req: Start request
    Note over dep_req: Run code up to yield
    dep_req ->> dep_func: Pass dependency
    Note over dep_func: Run code up to yield
    dep_func ->> operation: Run path operation with dependency
    operation ->> dep_func: Return from path operation
    Note over dep_func: Run code after yield
    Note over dep_func: ✅ Dependency closed
    dep_func ->> client: Send response to client
    Note over client: Response sent
    Note over dep_req: Run code after yield
    Note over dep_req: ✅ Dependency closed

yield, HTTPException, except 및 백그라운드 작업을 사용하는 의존성

yield를 사용하는 의존성은 시간이 지나면서 서로 다른 사용 사례를 다루고 일부 문제를 수정하기 위해 발전해 왔습니다.

FastAPI의 여러 버전에서 무엇이 바뀌었는지 보고 싶다면, 고급 가이드의 고급 의존성 - yield, HTTPException, except 및 백그라운드 작업을 사용하는 의존성에서 더 자세히 읽을 수 있습니다.

컨텍스트 관리자

"컨텍스트 관리자"란

"컨텍스트 관리자"는 Python에서 with 문에서 사용할 수 있는 모든 객체를 의미합니다.

예를 들어, with를 사용하여 파일을 읽을 수 있습니다:

with open("./somefile.txt") as f:
    contents = f.read()
    print(contents)

내부적으로 open("./somefile.txt") 는 "컨텍스트 관리자(Context Manager)"라고 불리는 객체를 생성합니다.

with 블록이 끝나면, 예외가 발생했더라도 파일을 닫도록 보장합니다.

yield가 있는 의존성을 생성하면 FastAPI는 내부적으로 이를 위한 컨텍스트 매니저를 생성하고 다른 관련 도구들과 결합합니다.

yield를 사용하는 의존성에서 컨텍스트 관리자 사용하기

경고

이것은 어느 정도 "고급" 개념입니다.

FastAPI를 처음 시작하는 경우 지금은 이 부분을 건너뛰어도 좋습니다.

Python에서는 다음을 통해 컨텍스트 관리자를 생성할 수 있습니다. 두 가지 메서드가 있는 클래스를 생성합니다: __enter__() and __exit__().

FastAPIyield가 있는 의존성 내에서 with 또는 async with문을 사용하여 이들을 활용할 수 있습니다:

class MySuperContextManager:
    def __init__(self):
        self.db = DBSession()

    def __enter__(self):
        return self.db

    def __exit__(self, exc_type, exc_value, traceback):
        self.db.close()


async def get_db():
    with MySuperContextManager() as db:
        yield db

컨텍스트 관리자를 생성하는 또 다른 방법은 다음과 같습니다:

이들은 단일 yield가 있는 함수를 꾸미는 데 사용합니다.

이것이 FastAPIyield가 있는 의존성을 위해 내부적으로 사용하는 방식입니다.

하지만 FastAPI 의존성에는 이러한 데코레이터를 사용할 필요가 없습니다(그리고 사용해서도 안됩니다).

FastAPI가 내부적으로 이를 처리해 줄 것입니다.