Dependencias Avanzadas¶
Dependencias parametrizadas¶
Todas las dependencias que hemos visto son una función o clase fija.
Pero podría haber casos donde quieras poder establecer parámetros en la dependencia, sin tener que declarar muchas funciones o clases diferentes.
Imaginemos que queremos tener una dependencia que verifique si el query parameter q contiene algún contenido fijo.
Pero queremos poder parametrizar ese contenido fijo.
Una instancia "callable"¶
En Python hay una forma de hacer que una instancia de una clase sea "callable".
No la clase misma (que ya es un callable), sino una instancia de esa clase.
Para hacer eso, declaramos un método __call__:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
return {"fixed_content_in_query": fixed_content_included}
🤓 Otras versiones y variantes
Consejo
Preferible usar la versión con Annotated si es posible.
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
return {"fixed_content_in_query": fixed_content_included}
En este caso, este __call__ es lo que FastAPI usará para verificar parámetros adicionales y sub-dependencias, y esto es lo que se llamará para pasar un valor al parámetro en tu path operation function más adelante.
Parametrizar la instancia¶
Y ahora, podemos usar __init__ para declarar los parámetros de la instancia que podemos usar para "parametrizar" la dependencia:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
return {"fixed_content_in_query": fixed_content_included}
🤓 Otras versiones y variantes
Consejo
Preferible usar la versión con Annotated si es posible.
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
return {"fixed_content_in_query": fixed_content_included}
En este caso, FastAPI nunca va a tocar ni a preocuparse por __init__, lo usaremos directamente en nuestro código.
Crear una instancia¶
Podríamos crear una instancia de esta clase con:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
return {"fixed_content_in_query": fixed_content_included}
🤓 Otras versiones y variantes
Consejo
Preferible usar la versión con Annotated si es posible.
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
return {"fixed_content_in_query": fixed_content_included}
Y de esa manera podemos "parametrizar" nuestra dependencia, que ahora tiene "bar" dentro, como el atributo checker.fixed_content.
Usar la instancia como dependencia¶
Luego, podríamos usar este checker en un Depends(checker), en lugar de Depends(FixedContentQueryChecker), porque la dependencia es la instancia, checker, no la clase misma.
Y al resolver la dependencia, FastAPI llamará a este checker así:
checker(q="somequery")
...y pasar lo que devuelva como el valor de la dependencia en nuestra path operation function como el parámetro fixed_content_included:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
return {"fixed_content_in_query": fixed_content_included}
🤓 Otras versiones y variantes
Consejo
Preferible usar la versión con Annotated si es posible.
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
return {"fixed_content_in_query": fixed_content_included}
Consejo
Todo esto podría parecer artificial. Y podría no estar muy claro cómo es útil todavía.
Estos ejemplos son intencionalmente simples, pero muestran cómo funciona todo.
En los capítulos sobre seguridad, hay funciones de utilidad que están implementadas de esta misma manera.
Si entendiste todo esto, ya sabes cómo funcionan esas herramientas de utilidad para seguridad por debajo.
Dependencias con yield, HTTPException, except y Background Tasks¶
Aviso
Muy probablemente no necesites estos detalles técnicos.
Estos detalles son útiles principalmente si tenías una aplicación FastAPI anterior a 0.121.0 y estás enfrentando problemas con dependencias con yield.
Las dependencias con yield han evolucionado con el tiempo para abarcar los diferentes casos de uso y para corregir algunos problemas, aquí hay un resumen de lo que ha cambiado.
Dependencias con yield y scope¶
En la versión 0.121.0, FastAPI añadió soporte para Depends(scope="function") para dependencias con yield.
Usando Depends(scope="function"), el exit code después de yield se ejecuta justo después de que la path operation function termina, antes de que la respuesta sea enviada al cliente.
Y cuando se usa Depends(scope="request") (el valor por defecto), el exit code después de yield se ejecuta después de que la respuesta es enviada.
Puedes leer más sobre esto en la documentación de Dependencies with yield - Early exit and scope.
Dependencias con yield y StreamingResponse, Detalles Técnicos¶
Antes de FastAPI 0.118.0, si usabas una dependencia con yield, ejecutaría el exit code después de que la path operation function retornara pero justo antes de enviar la respuesta.
La intención era evitar mantener recursos por más tiempo del necesario, esperando a que la respuesta viajara por la red.
Este cambio también significó que si devolvías un StreamingResponse, el exit code de la dependencia con yield ya habría sido ejecutado.
Por ejemplo, si tenías una sesión de base de datos en una dependencia con yield, el StreamingResponse no podría usar esa sesión mientras transmite datos porque la sesión ya habría sido cerrada en el exit code después de yield.
Este comportamiento se revirtió en 0.118.0, para hacer que el exit code después de yield se ejecute después de que la respuesta sea enviada.
Nota
Como verás más abajo, esto es muy similar al comportamiento anterior a la versión 0.106.0, pero con varias mejoras y correcciones de errores para casos límite.
Casos de Uso con Early Exit Code¶
Hay algunos casos de uso con condiciones específicas que podrían beneficiarse del comportamiento anterior de ejecutar el exit code de las dependencias con yield antes de enviar la respuesta.
Por ejemplo, imagina que tienes código que usa una sesión de base de datos en una dependencia con yield solo para verificar un usuario, pero la sesión de base de datos nunca se usa nuevamente en la path operation function, solo en la dependencia, y la respuesta tarda mucho en enviarse, como un StreamingResponse que envía datos lentamente, pero por alguna razón no usa la base de datos.
En este caso, la sesión de la base de datos se mantendría hasta que la respuesta termine de enviarse, pero si no la usas, entonces no sería necesario mantenerla.
Así podría verse:
import time
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine
engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
app = FastAPI()
def get_session():
with Session(engine) as session:
yield session
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=403, detail="Not authorized")
def generate_stream(query: str):
for ch in query:
yield ch
time.sleep(0.1)
@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
return StreamingResponse(content=generate_stream(query))
El exit code, el cierre automático de la Session en:
# Code above omitted 👆
def get_session():
with Session(engine) as session:
yield session
# Code below omitted 👇
👀 Vista previa del archivo completo
import time
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine
engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
app = FastAPI()
def get_session():
with Session(engine) as session:
yield session
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=403, detail="Not authorized")
def generate_stream(query: str):
for ch in query:
yield ch
time.sleep(0.1)
@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
return StreamingResponse(content=generate_stream(query))
...se ejecutaría después de que la respuesta termine de enviar los datos lentos:
# Code above omitted 👆
def generate_stream(query: str):
for ch in query:
yield ch
time.sleep(0.1)
@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
return StreamingResponse(content=generate_stream(query))
👀 Vista previa del archivo completo
import time
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine
engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
app = FastAPI()
def get_session():
with Session(engine) as session:
yield session
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=403, detail="Not authorized")
def generate_stream(query: str):
for ch in query:
yield ch
time.sleep(0.1)
@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
return StreamingResponse(content=generate_stream(query))
Pero como generate_stream() no usa la sesión de la base de datos, no es realmente necesario mantener la sesión abierta mientras se envía la respuesta.
Si tienes este caso de uso específico usando SQLModel (o SQLAlchemy), podrías cerrar explícitamente la sesión después de que ya no la necesites:
# Code above omitted 👆
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=403, detail="Not authorized")
session.close()
# Code below omitted 👇
👀 Vista previa del archivo completo
import time
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine
engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
app = FastAPI()
def get_session():
with Session(engine) as session:
yield session
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=403, detail="Not authorized")
session.close()
def generate_stream(query: str):
for ch in query:
yield ch
time.sleep(0.1)
@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
return StreamingResponse(content=generate_stream(query))
De esa manera la sesión liberaría la conexión a la base de datos, para que otras requests pudieran usarla.
Si tienes un caso de uso diferente que necesita salir temprano de una dependencia con yield, por favor crea un GitHub Discussion Question con tu caso de uso específico y por qué te beneficiaría tener un cierre temprano para dependencias con yield.
Si hay casos de uso convincentes para el cierre temprano en dependencias con yield, consideraría añadir una nueva forma de optar por el cierre temprano.
Dependencias con yield y except, Detalles Técnicos¶
Antes de FastAPI 0.110.0, si usabas una dependencia con yield, y luego capturabas una excepción con except en esa dependencia, y no volvías a lanzar la excepción, la excepción sería automáticamente lanzada/reenviada a cualquier exception handler o al handler de error interno del servidor.
Esto se cambió en la versión 0.110.0 para corregir el consumo de memoria no manejado por excepciones reenviadas sin un handler (errores internos del servidor), y para hacerlo consistente con el comportamiento del código Python regular.
Background Tasks y Dependencias con yield, Detalles Técnicos¶
Antes de FastAPI 0.106.0, lanzar excepciones después de yield no era posible, el exit code en dependencias con yield se ejecutaba después de que la respuesta fuera enviada, así que los Exception Handlers ya habrían sido ejecutados.
Esto se diseñó así principalmente para permitir usar los mismos objetos "yielded" por las dependencias dentro de las background tasks, porque el exit code se ejecutaría después de que las background tasks terminaran.
Esto se cambió en FastAPI 0.106.0 con la intención de no mantener recursos mientras se espera a que la respuesta viaje por la red.
Consejo
Además, una background task es normalmente un conjunto independiente de lógica que debería manejarse por separado, con sus propios recursos (ej. su propia conexión a la base de datos).
Así, de esta manera probablemente tendrás código más limpio.
Si solías depender de este comportamiento, ahora deberías crear los recursos para las background tasks dentro de la background task misma, y usar internamente solo datos que no dependan de los recursos de las dependencias con yield.
Por ejemplo, en lugar de usar la misma sesión de base de datos, crearías una nueva sesión de base de datos dentro de la background task, y obtendrías los objetos de la base de datos usando esta nueva sesión. Y luego, en lugar de pasar el objeto de la base de datos como parámetro a la función de la background task, pasarías el ID de ese objeto y luego obtendrías el objeto nuevamente dentro de la función de la background task.