Ir para o conteúdo

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 utilizar yield apenas uma vez por dependência.

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.

Apenas o código anterior à 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:

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, 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 outra exceção, 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)
🤓 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)

E todas elas podem utilizar yield.

Neste caso, dependency_c, para executar seu código de saída, precisa que o valor de dependency_b (nomeado de dep_b aqui) continue disponível.

E, por outro lado, dependency_b precisa que o valor de dependency_a (nomeado de dep_a) esteja 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)
🤓 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)

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 das duas.

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 pode usar dependências com yield e ter blocos try que tentam executar algum código e depois executar algum código de saída com finally.

Você também pode usar except para capturar a exceção que foi levantada e fazer algo com ela.

Por exemplo, você pode levantar uma exceção diferente, como HTTPException.

Dica

Essa é uma técnica relativamente avançada, e na maioria dos casos você não vai precisar dela, já que você pode levantar exceções (incluindo HTTPException) dentro do resto do código da sua aplicação, por exemplo, na 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
🤓 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

Se você quiser capturar exceções e criar uma resposta personalizada com base nisso, crie 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 identificar 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
🤓 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

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) 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ê deve 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
🤓 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

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

Se você levantar qualquer exceção no código da função de operação de rota, ela será passada para as dependências com yield, incluindo 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.

Saída antecipada e scope

Normalmente, o código de saída das dependências com yield é executado após a resposta ser enviada ao cliente.

Mas se você sabe que não precisará usar a dependência depois de retornar da função de operação de rota, você pode usar Depends(scope="function") para dizer ao FastAPI que deve fechar a dependência depois que a função de operação de rota retornar, mas antes de a resposta ser enviada.

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 
from fastapi import Depends, FastAPI
from typing_extensions import Annotated

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

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() recebe um parâmetro scope que pode ser:

  • "function": iniciar a dependência antes da função de operação de rota que trata a requisição, encerrar a dependência depois que a função de operação de rota termina, mas antes de a resposta ser enviada de volta ao cliente. Assim, a função da dependência será executada em torno da função de operação de rota.
  • "request": iniciar a dependência antes da função de operação de rota que trata a requisição (semelhante a quando se usa "function"), mas encerrar depois que a resposta é enviada de volta ao cliente. Assim, a função da dependência será executada em torno do ciclo de requisição e resposta.

Se não for especificado e a dependência tiver yield, ela terá scope igual a "request" por padrão.

scope para subdependências

Quando você declara uma dependência com scope="request" (o padrão), qualquer subdependência também precisa ter scope igual a "request".

Mas uma dependência com scope igual a "function" pode ter dependências com scope igual a "function" e com scope igual a "request".

Isso porque qualquer dependência precisa conseguir executar seu código de saída antes das subdependências, pois pode ainda precisar usá-las durante seu código de saída.

sequenceDiagram

participant client as Cliente
participant dep_req as Dep scope="request"
participant dep_func as Dep scope="function"
participant operation as Operação de Rota

    client ->> dep_req: Iniciar requisição
    Note over dep_req: Executar código até o yield
    dep_req ->> dep_func: Passar dependência
    Note over dep_func: Executar código até o yield
    dep_func ->> operation: Executar operação de rota com dependência
    operation ->> dep_func: Retornar da operação de rota
    Note over dep_func: Executar código após o yield
    Note over dep_func: ✅ Dependência fechada
    dep_func ->> client: Enviar resposta ao cliente
    Note over client: Resposta enviada
    Note over dep_req: Executar código após o yield
    Note over dep_req: ✅ Dependência fechada

Dependências com yield, HTTPException, except e Tarefas de Background

Dependências com yield evoluíram ao longo do tempo para cobrir diferentes casos de uso e corrigir alguns problemas.

Se você quiser ver o que mudou em diferentes versões do FastAPI, você pode ler mais sobre isso no guia avançado, em Dependências Avançadas - Dependências com yield, HTTPException, except e Tarefas 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

Atenção

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.