Dependencias Avanzadas¶

Dependencias con parámetros¶

Todas las dependencias que hemos visto son una función o clase fija.

Pero podría haber casos en los que quieras poder establecer parámetros en la dependencia, sin tener que declarar muchas funciones o clases diferentes.

Imaginemos que queremos tener una dependencia que revise si el parámetro de queryq contiene algún contenido fijo.

Pero queremos poder parametrizar ese contenido fijo.

Unainstance "callable"¶

En Python hay una forma de hacer que una instance de una clase sea un "callable".

No la clase en sí (que ya es un callable), sino una instance de esa clase.

Para hacer eso, declaramos un método__call__:

fromtypingimportAnnotatedfromfastapiimportDepends,FastAPIapp=FastAPI()classFixedContentQueryChecker:def__init__(self,fixed_content:str):self.fixed_content=fixed_contentdef__call__(self,q:str=""):ifq:returnself.fixed_contentinqreturnFalsechecker=FixedContentQueryChecker("bar")@app.get("/query-checker/")asyncdefread_query_check(fixed_content_included:Annotated[bool,Depends(checker)]):return{"fixed_content_in_query":fixed_content_included}

fromfastapiimportDepends,FastAPIapp=FastAPI()classFixedContentQueryChecker:def__init__(self,fixed_content:str):self.fixed_content=fixed_contentdef__call__(self,q:str=""):ifq:returnself.fixed_contentinqreturnFalsechecker=FixedContentQueryChecker("bar")@app.get("/query-checker/")asyncdefread_query_check(fixed_content_included:bool=Depends(checker)):return{"fixed_content_in_query":fixed_content_included}

En este caso, este__call__ es lo queFastAPI usará para comprobar parámetros adicionales y sub-dependencias, y es lo que llamará para pasar un valor al parámetro en tupath operation function más adelante.

Parametrizar la instance¶

Y ahora, podemos usar__init__ para declarar los parámetros de la instance que podemos usar para "parametrizar" la dependencia:

fromtypingimportAnnotatedfromfastapiimportDepends,FastAPIapp=FastAPI()classFixedContentQueryChecker:def__init__(self,fixed_content:str):self.fixed_content=fixed_contentdef__call__(self,q:str=""):ifq:returnself.fixed_contentinqreturnFalsechecker=FixedContentQueryChecker("bar")@app.get("/query-checker/")asyncdefread_query_check(fixed_content_included:Annotated[bool,Depends(checker)]):return{"fixed_content_in_query":fixed_content_included}

fromfastapiimportDepends,FastAPIapp=FastAPI()classFixedContentQueryChecker:def__init__(self,fixed_content:str):self.fixed_content=fixed_contentdef__call__(self,q:str=""):ifq:returnself.fixed_contentinqreturnFalsechecker=FixedContentQueryChecker("bar")@app.get("/query-checker/")asyncdefread_query_check(fixed_content_included:bool=Depends(checker)):return{"fixed_content_in_query":fixed_content_included}

En este caso,FastAPI nunca tocará ni se preocupará por__init__, lo usaremos directamente en nuestro código.

Crear una instance¶

Podríamos crear una instance de esta clase con:

fromtypingimportAnnotatedfromfastapiimportDepends,FastAPIapp=FastAPI()classFixedContentQueryChecker:def__init__(self,fixed_content:str):self.fixed_content=fixed_contentdef__call__(self,q:str=""):ifq:returnself.fixed_contentinqreturnFalsechecker=FixedContentQueryChecker("bar")@app.get("/query-checker/")asyncdefread_query_check(fixed_content_included:Annotated[bool,Depends(checker)]):return{"fixed_content_in_query":fixed_content_included}

fromfastapiimportDepends,FastAPIapp=FastAPI()classFixedContentQueryChecker:def__init__(self,fixed_content:str):self.fixed_content=fixed_contentdef__call__(self,q:str=""):ifq:returnself.fixed_contentinqreturnFalsechecker=FixedContentQueryChecker("bar")@app.get("/query-checker/")asyncdefread_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 de ella, como el atributochecker.fixed_content.

Usar la instance como una dependencia¶

Luego, podríamos usar estechecker en unDepends(checker), en lugar deDepends(FixedContentQueryChecker), porque la dependencia es la instance,checker, no la clase en sí.

Y al resolver la dependencia,FastAPI llamará a estechecker así:

checker(q="somequery")

...y pasará lo que eso retorne como el valor de la dependencia en nuestrapath operation function como el parámetrofixed_content_included:

fromtypingimportAnnotatedfromfastapiimportDepends,FastAPIapp=FastAPI()classFixedContentQueryChecker:def__init__(self,fixed_content:str):self.fixed_content=fixed_contentdef__call__(self,q:str=""):ifq:returnself.fixed_contentinqreturnFalsechecker=FixedContentQueryChecker("bar")@app.get("/query-checker/")asyncdefread_query_check(fixed_content_included:Annotated[bool,Depends(checker)]):return{"fixed_content_in_query":fixed_content_included}

fromfastapiimportDepends,FastAPIapp=FastAPI()classFixedContentQueryChecker:def__init__(self,fixed_content:str):self.fixed_content=fixed_contentdef__call__(self,q:str=""):ifq:returnself.fixed_contentinqreturnFalsechecker=FixedContentQueryChecker("bar")@app.get("/query-checker/")asyncdefread_query_check(fixed_content_included:bool=Depends(checker)):return{"fixed_content_in_query":fixed_content_included}

Dependencias conyield,HTTPException,except y Tareas en segundo plano¶

Las dependencias conyield han evolucionado con el tiempo para cubrir diferentes casos de uso y arreglar algunos problemas; aquí tienes un resumen de lo que ha cambiado.

Dependencias conyield yscope¶

En la versión 0.121.0, FastAPI agregó soporte paraDepends(scope="function") para dependencias conyield.

UsandoDepends(scope="function"), el código de salida después deyield se ejecuta justo después de que lapath operation function termina, antes de que la response se envíe de vuelta al cliente.

Y al usarDepends(scope="request") (el valor por defecto), el código de salida después deyield se ejecuta después de que la response es enviada.

Puedes leer más al respecto en la documentación deDependencias conyield - Salida temprana yscope.

Dependencias conyield yStreamingResponse, detalles técnicos¶

Antes de FastAPI 0.118.0, si usabas una dependencia conyield, ejecutaba el código de salida después de que lapath operation function retornaba pero justo antes de enviar la response.

La intención era evitar retener recursos por más tiempo del necesario, esperando a que la response viajara por la red.

Este cambio también significaba que si retornabas unStreamingResponse, el código de salida de la dependencia conyield ya se habría ejecutado.

Por ejemplo, si tenías una sesión de base de datos en una dependencia conyield, elStreamingResponse no podría usar esa sesión mientras hace streaming de datos porque la sesión ya se habría cerrado en el código de salida después deyield.

Este comportamiento se revirtió en la 0.118.0, para hacer que el código de salida después deyield se ejecute después de que la response sea enviada.

Casos de uso con salida temprana del código¶

Hay algunos casos de uso con condiciones específicas que podrían beneficiarse del comportamiento antiguo de ejecutar el código de salida de dependencias conyield antes de enviar la response.

Por ejemplo, imagina que tienes código que usa una sesión de base de datos en una dependencia conyield solo para verificar un usuario, pero la sesión de base de datos no se vuelve a usar en lapath operation function, solo en la dependencia, y la response tarda mucho en enviarse, como unStreamingResponse que envía datos lentamente, pero que por alguna razón no usa la base de datos.

En este caso, la sesión de base de datos se mantendría hasta que la response termine de enviarse, pero si no la usas, entonces no sería necesario mantenerla.

Así es como se vería:

importtimefromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPExceptionfromfastapi.responsesimportStreamingResponsefromsqlmodelimportField,Session,SQLModel,create_engineengine=create_engine("postgresql+psycopg://postgres:postgres@localhost/db")classUser(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:strapp=FastAPI()defget_session():withSession(engine)assession:yieldsessiondefget_user(user_id:int,session:Annotated[Session,Depends(get_session)]):user=session.get(User,user_id)ifnotuser:raiseHTTPException(status_code=403,detail="Not authorized")defgenerate_stream(query:str):forchinquery:yieldchtime.sleep(0.1)@app.get("/generate",dependencies=[Depends(get_user)])defgenerate(query:str):returnStreamingResponse(content=generate_stream(query))

El código de salida, el cierre automático de laSession en:

# Code above omitted 👆defget_session():withSession(engine)assession:yieldsession# Code below omitted 👇

importtimefromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPExceptionfromfastapi.responsesimportStreamingResponsefromsqlmodelimportField,Session,SQLModel,create_engineengine=create_engine("postgresql+psycopg://postgres:postgres@localhost/db")classUser(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:strapp=FastAPI()defget_session():withSession(engine)assession:yieldsessiondefget_user(user_id:int,session:Annotated[Session,Depends(get_session)]):user=session.get(User,user_id)ifnotuser:raiseHTTPException(status_code=403,detail="Not authorized")defgenerate_stream(query:str):forchinquery:yieldchtime.sleep(0.1)@app.get("/generate",dependencies=[Depends(get_user)])defgenerate(query:str):returnStreamingResponse(content=generate_stream(query))

...se ejecutaría después de que la response termine de enviar los datos lentos:

# Code above omitted 👆defgenerate_stream(query:str):forchinquery:yieldchtime.sleep(0.1)@app.get("/generate",dependencies=[Depends(get_user)])defgenerate(query:str):returnStreamingResponse(content=generate_stream(query))

importtimefromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPExceptionfromfastapi.responsesimportStreamingResponsefromsqlmodelimportField,Session,SQLModel,create_engineengine=create_engine("postgresql+psycopg://postgres:postgres@localhost/db")classUser(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:strapp=FastAPI()defget_session():withSession(engine)assession:yieldsessiondefget_user(user_id:int,session:Annotated[Session,Depends(get_session)]):user=session.get(User,user_id)ifnotuser:raiseHTTPException(status_code=403,detail="Not authorized")defgenerate_stream(query:str):forchinquery:yieldchtime.sleep(0.1)@app.get("/generate",dependencies=[Depends(get_user)])defgenerate(query:str):returnStreamingResponse(content=generate_stream(query))

Pero comogenerate_stream() no usa la sesión de base de datos, no es realmente necesario mantener la sesión abierta mientras se envía la response.

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 👆defget_user(user_id:int,session:Annotated[Session,Depends(get_session)]):user=session.get(User,user_id)ifnotuser:raiseHTTPException(status_code=403,detail="Not authorized")session.close()# Code below omitted 👇

importtimefromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPExceptionfromfastapi.responsesimportStreamingResponsefromsqlmodelimportField,Session,SQLModel,create_engineengine=create_engine("postgresql+psycopg://postgres:postgres@localhost/db")classUser(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:strapp=FastAPI()defget_session():withSession(engine)assession:yieldsessiondefget_user(user_id:int,session:Annotated[Session,Depends(get_session)]):user=session.get(User,user_id)ifnotuser:raiseHTTPException(status_code=403,detail="Not authorized")session.close()defgenerate_stream(query:str):forchinquery:yieldchtime.sleep(0.1)@app.get("/generate",dependencies=[Depends(get_user)])defgenerate(query:str):returnStreamingResponse(content=generate_stream(query))

De esa manera la sesión liberaría la conexión a la base de datos, para que otras requests puedan usarla.

Si tienes un caso de uso diferente que necesite salir temprano desde una dependencia conyield, por favor crea unaPregunta de Discusión en GitHub con tu caso de uso específico y por qué te beneficiaría tener cierre temprano para dependencias conyield.

Si hay casos de uso convincentes para el cierre temprano en dependencias conyield, consideraría agregar una nueva forma de optar por el cierre temprano.

Dependencias conyield yexcept, detalles técnicos¶

Antes de FastAPI 0.110.0, si usabas una dependencia conyield, y luego capturabas una excepción conexcept en esa dependencia, y no volvías a elevar la excepción, la excepción se elevaría/remitiría automáticamente a cualquier manejador de excepciones o al manejador de error interno del servidor.

Esto cambió en la versión 0.110.0 para arreglar consumo de memoria no manejado por excepciones reenviadas sin un manejador (errores internos del servidor), y para hacerlo consistente con el comportamiento del código Python normal.

Tareas en segundo plano y dependencias conyield, detalles técnicos¶

Antes de FastAPI 0.106.0, elevar excepciones después deyield no era posible, el código de salida en dependencias conyield se ejecutaba después de que la response era enviada, por lo queManejadores de Excepciones ya habrían corrido.

Esto se diseñó así principalmente para permitir usar los mismos objetos devueltos conyield por las dependencias dentro de tareas en segundo plano, porque el código de salida se ejecutaría después de que las tareas en segundo plano terminaran.

Esto cambió en FastAPI 0.106.0 con la intención de no retener recursos mientras se espera a que la response viaje por la red.

Si solías depender de este comportamiento, ahora deberías crear los recursos para las tareas en segundo plano dentro de la propia tarea en segundo plano, y usar internamente solo datos que no dependan de los recursos de dependencias conyield.

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 tarea en segundo plano, y obtendrías los objetos de la base de datos usando esta nueva sesión. Y entonces, en lugar de pasar el objeto de la base de datos como parámetro a la función de la tarea en segundo plano, pasarías el ID de ese objeto y luego obtendrías el objeto de nuevo dentro de la función de la tarea en segundo plano.