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는 일반 의존성과 마찬가지로 각각의 함수를 올바르게 처리할 것입니다.
yield와 try를 사용하는 의존성¶
yield를 사용하는 의존성에서 try 블록을 사용한다면, 의존성을 사용하는 도중 발생한 모든 예외를 받을 수 있습니다.
예를 들어, 다른 의존성이나 경로 작업의 중간에 데이터베이스 트랜잭션 "롤백"이 발생하거나 다른 오류가 발생한다면, 해당 예외를 의존성에서 받을 수 있습니다.
따라서, 의존성 내에서 except SomeException을 사용하여 특정 예외를 처리할 수 있습니다.
마찬가지로, finally를 사용하여 예외 발생 여부와 관계 없이 종료 단계까 실행되도록 할 수 있습니다.
async def get_db():
db = DBSession()
try:
yield db
finally:
db.close()
yield를 사용하는 하위 의존성¶
모든 크기와 형태의 하위 의존성과 하위 의존성의 "트리"도 가질 수 있으며, 이들 모두가 yield를 사용할 수 있습니다.
FastAPI는 yield를 사용하는 각 의존성의 "종료 코드"가 올바른 순서로 실행되도록 보장합니다.
예를 들어, dependency_c는 dependency_b에 의존할 수 있고, dependency_b는 dependency_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
from fastapi import Depends
from typing_extensions import Annotated
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)
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
from fastapi import Depends
from typing_extensions import Annotated
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)
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는 모든 것이 올바른 순서로 실행되도록 보장합니다.
yield와 HTTPException를 사용하는 의존성¶
yield와 try 블록이 있는 의존성을 사용하여 예외를 처리할 수 있다는 것을 알게 되었습니다.
같은 방식으로, yield 이후의 종료 코드에서 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
from fastapi import Depends, FastAPI, HTTPException
from typing_extensions import Annotated
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
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
예외를 처리하고(또는 추가로 다른 HTTPException을 발생시키기 위해) 사용할 수 있는 또 다른 방법은 사용자 정의 예외 처리기를 생성하는 것 입니다.
yield와 except를 사용하는 의존성¶
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
from fastapi import Depends, FastAPI, HTTPException
from typing_extensions import Annotated
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
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 응답을 보게 되지만, 서버는 어떤 오류가 발생했는지에 대한 로그나 다른 표시를 전혀 가지지 않게 됩니다. 😱
yield와 except를 사용하는 의존성에서 항상 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
from fastapi import Depends, FastAPI, HTTPException
from typing_extensions import Annotated
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
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를 사용하는 의존성에서 처리한 예외나 사용자 정의 예외처리기.를 사용하여 처리한 다른 예외도 발생시킬 수 있습니다.
어떤 예외가 발생하든, HTTPException을 포함하여 yield를 사용하는 의존성으로 전달됩니다. 대부분의 경우 예외를 다시 발생시키거나 새로운 예외를 발생시켜야 합니다.
yield, HTTPException, except 및 백그라운드 작업을 사용하는 의존성¶
경고
이러한 기술적 세부 사항은 대부분 필요하지 않으므로 이 섹션을 건너뛰고 아래에서 계속 진행해도 됩니다.
이러한 세부 정보는 주로 FastAPI 0.106.0 이전 버전에서 yield가 있는 의존성의 리소스를 백그라운드 작업에서 사용했던 경우메 유용합니다.
yield와 except를 사용하는 의존성, 기술 세부사항¶
FastAPI 0.110.0 이전에는 yield가 포함된 의존성을 사용한 후 해당 의존성에서 except가 포함된 예외를 캡처하고 다시 예외를 발생시키지 않으면 예외가 자동으로 예외 핸들러 또는 내부 서버 오류 핸들러로 발생/전달되었습니다.
이는 처리기 없이 전달된 예외(내부 서버 오류)에서 처리되지 않은 메모리 소비를 수정하고 일반 파이썬 코드의 동작과 일치하도록 하기 위해 0.110.0 버전에서 변경되었습니다.
백그라운드 작업과 yield를 사용하는 의존성, 기술 세부사항¶
FastAPI 0.106.0 이전에는 yield 이후에 예외를 발생시키는 것이 불가능했습니다. yield가 있는 의존성 종료 코드는 응답이 전송된 이후에 실행되었기 때문에, 예외 처리기가 이미 실행된 상태였습니다.
이는 주로 백그라운드 작업 내에서 의존성에서 "yield된" 동일한 객체를 사용할 수 있도록 하기 위해 이런 방식으로 설계되었습니다. 종료 코드는 백그라운드 작업이 완료된 후에 실행되었기 때문입니다
하지만 이렇게 하면 리소스를 불필요하게 양보한 의존성(예: 데이터베이스 연결)에서 보유하면서 응답이 네트워크를 통해 이동할 때까지 기다리는 것을 의미하기 때문에 FastAPI 0.106.0에서 변경되었습니다.
팁
또한 백그라운드 작업은 일반적으로 자체 리소스(예: 자체 데이터베이스 연결)를 사용하여 별도로 처리해야 하는 독립적인 로직 집합입니다.
따라서 이렇게 하면 코드가 더 깔끔해집니다.
만약 이전에 이러한 동작에 의존했다면, 이제는 백그라운드 작업 내부에서 백그라운드 작업을 위한 리소스를 생성하고, yield가 있는 의존성의 리소스에 의존하지 않는 데이터만 내부적으로 사용해야합니다.
예를 들어, 동일한 데이터베이스 세션을 사용하는 대신, 백그라운드 작업 내부에서 새로운 데이터베이스 세션을 생성하고 이 새로운 세션을 사용하여 데이터베이스에서 객체를 가져와야 합니다. 그리고 데이터베이스 객체를 백그라운드 작업 함수의 매개변수로 직접 전달하는 대신, 해당 객체의 ID를 전달한 다음 백그라운드 작업 함수 내부에서 객체를 다시 가져와야 합니다
컨텍스트 관리자¶
"컨텍스트 관리자"란?¶
"컨텍스트 관리자"는 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__().
FastAPI의 yield가 있는 의존성 내에서
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가 있는 함수를 꾸미는 데 사용합니다.
이것이 FastAPI가 yield가 있는 의존성을 위해 내부적으로 사용하는 방식입니다.
하지만 FastAPI 의존성에는 이러한 데코레이터를 사용할 필요가 없습니다(그리고 사용해서도 안됩니다).
FastAPI가 내부적으로 이를 처리해 줄 것입니다.