Dependencias con yield¶

FastAPI soporta dependencias que hacen algunos pasos extra después de terminar.

Para hacer esto, usa yield en lugar de return, y escribe los pasos extra (código) después.

Una dependencia de base de datos con yield¶

Por ejemplo, podrías usar esto para crear una sesión de base de datos y cerrarla después de terminar.

Solo el código previo e incluyendo el statement yield se ejecuta antes de crear una respuesta:

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

El valor yielded es lo que se inyecta en las path operations y otras dependencias:

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

El código que sigue al statement yield se ejecuta después de la respuesta:

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

Una dependencia con yield y try¶

Si usas un bloque try en una dependencia con yield, recibirás cualquier excepción que haya sido lanzada al usar la dependencia.

Por ejemplo, si algún código en algún punto intermedio, en otra dependencia o en una path operation, hizo un "rollback" de una transacción de base de datos o creó cualquier otra excepción, recibirías la excepción en tu dependencia.

Entonces, puedes buscar esa excepción específica dentro de la dependencia con except SomeException.

De la misma manera, puedes usar finally para asegurarte de que los pasos de salida se ejecuten, sin importar si hubo una excepción o no.

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

Sub-dependencias con yield¶

Puedes tener sub-dependencias y "árboles" de sub-dependencias de cualquier tamaño y forma, y cualquiera o todas ellas pueden usar yield.

FastAPI se asegurará de que el "exit code" en cada dependencia con yield se ejecute en el orden correcto.

Por ejemplo, dependency_c puede tener una dependencia en dependency_b, y dependency_b en 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


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)

Y todas ellas pueden usar yield.

En este caso dependency_c, para ejecutar su exit code, necesita que el valor de dependency_b (aquí llamado dep_b) siga disponible.

Y, a su vez, dependency_b necesita que el valor de dependency_a (aquí llamado dep_a) esté disponible para su exit code.

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


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)

De la misma manera, podrías tener algunas dependencias con yield y otras dependencias con return, y hacer que algunas de esas dependan de otras.

Y podrías tener una sola dependencia que requiere varias otras dependencias con yield, etc.

Puedes tener cualquier combinación de dependencias que quieras.

FastAPI se asegurará de que todo se ejecute en el orden correcto.

Dependencias con yield y HTTPException¶

Viste que puedes usar dependencias con yield y tener bloques try que intentan ejecutar algo de código y luego ejecutan algo de exit code después de finally.

También puedes usar except para capturar la excepción que fue lanzada y hacer algo con ella.

Por ejemplo, puedes lanzar una excepción diferente, como 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

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

Si quieres capturar excepciones y crear una respuesta personalizada basada en eso, crea un Custom Exception Handler.

Dependencias con yield y except¶

Si capturas una excepción usando except en una dependencia con yield y no la lanzas de nuevo (o lanzas una nueva excepción), FastAPI no podrá notar que hubo una excepción, de la misma manera que pasaría con Python regular:

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

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

En este caso, el cliente verá una respuesta HTTP 500 Internal Server Error como debería, dado que no estamos lanzando un HTTPException o similar, pero el servidor no tendrá ningún log ni ninguna otra indicación de cuál fue el error. 😱

Siempre raise en Dependencias con yield y except¶

Si capturas una excepción en una dependencia con yield, a menos que estés lanzando otro HTTPException o similar, deberías volver a lanzar la excepción original.

Puedes volver a lanzar la misma excepción usando 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

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

Ahora el cliente obtendrá la misma respuesta HTTP 500 Internal Server Error, pero el servidor tendrá nuestro InternalError personalizado en los logs. 😎

Ejecución de dependencias con yield¶

La secuencia de ejecución es más o menos como este diagrama. El tiempo fluye de arriba hacia abajo. Y cada columna es una de las partes interactuando o ejecutando código.

Salida temprana y scope¶

Normalmente el exit code de las dependencias con yield se ejecuta después de la respuesta que se envía al cliente.

Pero si sabes que no necesitarás usar la dependencia después de retornar de la path operation function, puedes usar Depends(scope="function") para decirle a FastAPI que debería cerrar la dependencia después de que la path operation function retorne, pero antes de que la respuesta sea 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

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() recibe un parámetro scope que puede ser:

• "function": inicia la dependencia antes de la path operation function que maneja la petición, termina la dependencia después de que la path operation function termine, pero antes de que la respuesta sea enviada de vuelta al cliente. Así, la función de dependencia será ejecutada alrededor de la path operation function.
• "request": inicia la dependencia antes de la path operation function que maneja la petición (similar a cuando se usa "function"), pero termina después de que la respuesta sea enviada de vuelta al cliente. Así, la función de dependencia será ejecutada alrededor del ciclo de petición y respuesta.

Si no se especifica y la dependencia tiene yield, tendrá un scope de "request" por defecto.

scope para sub-dependencias¶

Cuando declaras una dependencia con scope="request" (el valor por defecto), cualquier sub-dependencia necesita tener también un scope de "request".

Pero una dependencia con scope de "function" puede tener dependencias con scope de "function" y scope de "request".

Esto se debe a que cualquier dependencia necesita poder ejecutar su código de salida antes que las sub-dependencias, ya que podría necesitar usarlas durante su código de salida.

Dependencias con yield, HTTPException, except y Background Tasks¶

Las dependencias con yield han evolucionado con el tiempo para cubrir diferentes casos de uso y arreglar algunos problemas.

Si quieres ver qué ha cambiado en diferentes versiones de FastAPI, puedes leer más sobre esto en la guía avanzada, en Dependencias Avanzadas - Dependencias con yield, HTTPException, except y Background Tasks.

Context Managers¶

Qué son los "Context Managers"¶

Los "Context Managers" son cualquiera de esos objetos de Python que puedes usar en un statement with.

Por ejemplo, puedes usar with para leer un archivo:

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

Por debajo, open("./somefile.txt") crea un objeto que se llama "Context Manager".

Cuando el bloque with termina, se asegura de cerrar el archivo, incluso si hubo excepciones.

Cuando creas una dependencia con yield, FastAPI creará internamente un context manager para ella, y lo combinará con algunas otras herramientas relacionadas.

Usar context managers en dependencias con yield¶

En Python, puedes crear Context Managers creando una clase con dos métodos: __enter__() y __exit__().

También puedes usarlos dentro de dependencias de FastAPI con yield usando statements with o async with dentro de la función de dependencia:

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