Dependências com yield¶
O FastAPI possui suporte para dependências que realizam alguns passos extras ao finalizar.
Para fazer isso, utilize yield
em vez de return
, e escreva os passos extras (código) depois.
"Dica"
Garanta que yield
é utilizado apenas uma vez.
"Detalhes Técnicos"
Qualquer função que possa ser utilizada com:
pode ser utilizada como uma dependência do FastAPI.
Na realidade, o FastAPI utiliza esses dois decoradores internamente.
Uma dependência de banco de dados com yield
¶
Por exemplo, você poderia utilizar isso para criar uma sessão do banco de dados, e fechá-la após terminar sua operação.
Apenas o código anterior a declaração com yield
e o código contendo essa declaração são executados antes de criar uma resposta.
async def get_db():
db = DBSession()
try:
yield db
finally:
db.close()
O valor gerado (yielded) é o que é injetado nas operações de rota e outras dependências.
async def get_db():
db = DBSession()
try:
yield db
finally:
db.close()
O código após o yield
é executado após a resposta ser entregue:
async def get_db():
db = DBSession()
try:
yield db
finally:
db.close()
"Dica"
Você pode usar funções assíncronas (async
) ou funções comuns.
O FastAPI saberá o que fazer com cada uma, da mesma forma que as dependências comuns.
Uma dependência com yield
e try
¶
Se você utilizar um bloco try
em uma dependência com yield
, você irá capturar qualquer exceção que for lançada enquanto a dependência é utilizada.
Por exemplo, se algum código em um certo momento no meio da operação, em outra dependência ou em uma operação de rota, fizer um "rollback" de uma transação de banco de dados ou causar qualquer outro erro, você irá capturar a exceção em sua dependência.
Então, você pode procurar por essa exceção específica dentro da dependência com except AlgumaExcecao
.
Da mesma forma, você pode utilizar finally
para garantir que os passos de saída são executados, com ou sem exceções.
async def get_db():
db = DBSession()
try:
yield db
finally:
db.close()
Subdependências com yield
¶
Você pode ter subdependências e "árvores" de subdependências de qualquer tamanho e forma, e qualquer uma ou todas elas podem utilizar yield
.
O FastAPI garantirá que o "código de saída" em cada dependência com yield
é executado na ordem correta.
Por exemplo, dependency_c
pode depender de dependency_b
, e dependency_b
depender de 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)
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)
"Dica"
Utilize a versão com Annotated
se possível.
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)
E todas elas podem utilizar yield
.
Neste caso, dependency_c
precisa que o valor de dependency_b
(nomeada de dep_b
aqui) continue disponível para executar seu código de saída.
E, por outro lado, dependency_b
precisa que o valor de dependency_a
(nomeada de dep_a
) continue disponível para executar seu código de saída.
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)
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)
"Dica"
Utilize a versão com Annotated
se possível.
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)
Da mesma forma, você pode ter algumas dependências com yield
e outras com return
e ter uma relação de dependência entre algumas dos dois tipos.
E você poderia ter uma única dependência que precisa de diversas outras dependências com yield
, etc.
Você pode ter qualquer combinação de dependências que você quiser.
O FastAPI se encarrega de executá-las na ordem certa.
"Detalhes Técnicos"
Tudo isso funciona graças aos gerenciadores de contexto do Python.
O FastAPI utiliza eles internamente para alcançar isso.
Dependências com yield
e httpexception
¶
Você viu que dependências podem ser utilizadas com yield
e podem incluir blocos try
para capturar exceções.
Da mesma forma, você pode lançar uma httpexception
ou algo parecido no código de saída, após o yield
"Dica"
Essa é uma técnica relativamente avançada, e na maioria dos casos você não precisa dela totalmente, já que você pode lançar exceções (incluindo httpexception
) dentro do resto do código da sua aplicação, por exemplo, em uma função de operação de rota.
Mas ela existe para ser utilizada caso você precise. 🤓
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
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
"Dica"
Utilize a versão com Annotated
se possível.
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
Uma alternativa que você pode utilizar para capturar exceções (e possivelmente lançar outra HTTPException) é criar um Manipulador de Exceções Customizado.
Dependências com yield
e except
¶
Se você capturar uma exceção com except
em uma dependência que utilize yield
e ela não for levantada novamente (ou uma nova exceção for levantada), o FastAPI não será capaz de identifcar que houve uma exceção, da mesma forma que aconteceria com Python puro:
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
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
"dica"
utilize a versão com Annotated
se possível.
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
Neste caso, o cliente irá ver uma resposta HTTP 500 Internal Server Error como deveria acontecer, já que não estamos levantando nenhuma HTTPException
ou coisa parecida, mas o servidor não terá nenhum log ou qualquer outra indicação de qual foi o erro. 😱
Sempre levante (raise
) exceções em Dependências com yield
e except
¶
Se você capturar uma exceção em uma dependência com yield
, a menos que você esteja levantando outra HTTPException
ou coisa parecida, você deveria relançar a exceção original.
Você pode relançar a mesma exceção utilizando 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
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
"Dica"
Utilize a versão com Annotated
se possível.
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
Agora o cliente irá receber a mesma resposta HTTP 500 Internal Server Error, mas o servidor terá nosso InternalError
personalizado nos logs. 😎
Execução de dependências com yield
¶
A sequência de execução é mais ou menos como esse diagrama. O tempo passa do topo para baixo. E cada coluna é uma das partes interagindo ou executando código.
sequenceDiagram
participant client as Cliente
participant handler as Manipulador de exceções
participant dep as Dep com yield
participant operation as Operação de Rota
participant tasks as Tarefas de Background
Note over client,operation: pode lançar exceções, incluindo HTTPException
client ->> dep: Iniciar requisição
Note over dep: Executar código até o yield
opt lançar Exceção
dep -->> handler: lançar Exceção
handler -->> client: resposta de erro HTTP
end
dep ->> operation: Executar dependência, e.g. sessão de BD
opt raise
operation -->> dep: Lançar exceção (e.g. HTTPException)
opt handle
dep -->> dep: Pode capturar exceções, lançar uma nova HTTPException, lançar outras exceções
end
handler -->> client: resposta de erro HTTP
end
operation ->> client: Retornar resposta ao cliente
Note over client,operation: Resposta já foi enviada, e não pode ser modificada
opt Tarefas
operation -->> tasks: Enviar tarefas de background
end
opt Lançar outra exceção
tasks -->> tasks: Manipula exceções no código da tarefa de background
end
"Informação"
Apenas uma resposta será enviada para o cliente. Ela pode ser uma das respostas de erro, ou então a resposta da operação de rota.
Após uma dessas respostas ser enviada, nenhuma outra resposta pode ser enviada
"Dica"
Esse diagrama mostra HttpException
, mas você pode levantar qualquer outra exceção que você capture em uma dependência com yield
ou um Manipulador de exceções personalizado.
Se você lançar qualquer exceção, ela será passada para as dependências com yield, inlcuindo a HTTPException
. Na maioria dos casos você vai querer relançar essa mesma exceção ou uma nova a partir da dependência com yield
para garantir que ela seja tratada adequadamente.
Dependências com yield
, HTTPException
, except
e Tarefas de Background¶
"Aviso"
Você provavelmente não precisa desses detalhes técnicos, você pode pular essa seção e continuar na próxima seção abaixo.
Esses detalhes são úteis principalmente se você estiver usando uma versão do FastAPI anterior à 0.106.0 e utilizando recursos de dependências com yield
em tarefas de background.
Dependências com yield
e except
, Detalhes Técnicos¶
Antes do FastAPI 0.110.0, se você utilizasse uma dependência com yield
, e então capturasse uma dependência com except
nessa dependência, caso a exceção não fosse relançada, ela era automaticamente lançada para qualquer manipulador de exceções ou o manipulador de erros interno do servidor.
Isso foi modificado na versão 0.110.0 para consertar o consumo de memória não controlado das exceções relançadas automaticamente sem um manipulador (erros internos do servidor), e para manter o comportamento consistente com o código Python tradicional.
Tarefas de Background e Dependências com yield
, Detalhes Técnicos¶
Antes do FastAPI 0.106.0, levantar exceções após um yield
não era possível, o código de saída nas dependências com yield
era executado após a resposta ser enviada, então os Manipuladores de Exceções já teriam executado.
Isso foi implementado dessa forma principalmente para permitir que os mesmos objetos fornecidos ("yielded") pelas dependências dentro de tarefas de background fossem reutilizados, por que o código de saída era executado antes das tarefas de background serem finalizadas.
Ainda assim, como isso exigiria esperar que a resposta navegasse pela rede enquanto mantia ativo um recurso desnecessário na dependência com yield (por exemplo, uma conexão com banco de dados), isso mudou na versão 0.106.0 do FastAPI.
"Dica"
Adicionalmente, uma tarefa de background é, normalmente, um conjunto de lógicas independentes que devem ser manipuladas separadamente, com seus próprios recursos (e.g. sua própria conexão com banco de dados).
Então, dessa forma você provavelmente terá um código mais limpo.
Se você costumava depender desse comportamento, agora você precisa criar os recursos para uma tarefa de background dentro dela mesma, e usar internamente apenas dados que não dependam de recursos de dependências com yield
.
Por exemplo, em vez de utilizar a mesma sessão do banco de dados, você criaria uma nova sessão dentro da tarefa de background, e você obteria os objetos do banco de dados utilizando essa nova sessão. E então, em vez de passar o objeto obtido do banco de dados como um parâmetro para a função da tarefa de background, você passaria o ID desse objeto e buscaria ele novamente dentro da função da tarefa de background.
Gerenciadores de contexto¶
O que são gerenciadores de contexto¶
"Gerenciadores de Contexto" são qualquer um dos objetos Python que podem ser utilizados com a declaração with
.
Por exemplo, você pode utilizar with
para ler um arquivo:
with open("./somefile.txt") as f:
contents = f.read()
print(contents)
Por baixo dos panos, o código open("./somefile.txt")
cria um objeto que é chamado de "Gerenciador de Contexto".
Quando o bloco with
finaliza, ele se certifica de fechar o arquivo, mesmo que tenha ocorrido alguma exceção.
Quando você cria uma dependência com yield
, o FastAPI irá criar um gerenciador de contexto internamente para ela, e combiná-lo com algumas outras ferramentas relacionadas.
Utilizando gerenciadores de contexto em dependências com yield
¶
"Aviso"
Isso é uma ideia mais ou menos "avançada".
Se você está apenas iniciando com o FastAPI você pode querer pular isso por enquanto.
Em python, você pode criar Gerenciadores de Contexto ao criar uma classe com dois métodos: __enter__()
e __exit__()
.
Você também pode usá-los dentro de dependências com yield
do FastAPI ao utilizar with
ou async with
dentro da função da dependência:
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
"Dica"
Outra forma de criar um gerenciador de contexto é utilizando:
Para decorar uma função com um único yield
.
Isso é o que o FastAPI usa internamente para dependências com yield
.
Mas você não precisa usar esses decoradores para as dependências do FastAPI (e você não deveria).
O FastAPI irá fazer isso para você internamente.