Advanced Dependencies¶

Parameterized dependencies¶

All the dependencies we have seen are a fixed function or class.

But there could be cases where you want to be able to set parameters on the dependency, without having to declare many different functions or classes.

Let's imagine that we want to have a dependency that checks if the query parameterq contains some fixed content.

But we want to be able to parameterize that fixed content.

A "callable" instance¶

In Python there's a way to make an instance of a class a "callable".

Not the class itself (which is already a callable), but an instance of that class.

To do that, we declare a method__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}

In this case, this__call__ is whatFastAPI will use to check for additional parameters and sub-dependencies, and this is what will be called to pass a value to the parameter in yourpath operation function later.

Parameterize the instance¶

And now, we can use__init__ to declare the parameters of the instance that we can use to "parameterize" the dependency:

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}

In this case,FastAPI won't ever touch or care about__init__, we will use it directly in our code.

Create an instance¶

We could create an instance of this class with:

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}

And that way we are able to "parameterize" our dependency, that now has"bar" inside of it, as the attributechecker.fixed_content.

Use the instance as a dependency¶

Then, we could use thischecker in aDepends(checker), instead ofDepends(FixedContentQueryChecker), because the dependency is the instance,checker, not the class itself.

And when solving the dependency,FastAPI will call thischecker like:

checker(q="somequery")

...and pass whatever that returns as the value of the dependency in ourpath operation function as the parameterfixed_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}

Dependencies withyield,HTTPException,except and Background Tasks¶

Dependencies withyield have evolved over time to account for the different use cases and to fix some issues, here's a summary of what has changed.

Dependencies withyield andscope¶

In version 0.121.0, FastAPI added support forDepends(scope="function") for dependencies withyield.

UsingDepends(scope="function"), the exit code afteryield is executed right after thepath operation function is finished, before the response is sent back to the client.

And when usingDepends(scope="request") (the default), the exit code afteryield is executed after the response is sent.

You can read more about it in the docs forDependencies withyield - Early exit andscope.

Dependencies withyield andStreamingResponse, Technical Details¶

Before FastAPI 0.118.0, if you used a dependency withyield, it would run the exit code after thepath operation function returned but right before sending the response.

The intention was to avoid holding resources for longer than necessary, waiting for the response to travel through the network.

This change also meant that if you returned aStreamingResponse, the exit code of the dependency withyield would have been already run.

For example, if you had a database session in a dependency withyield, theStreamingResponse would not be able to use that session while streaming data because the session would have already been closed in the exit code afteryield.

This behavior was reverted in 0.118.0, to make the exit code afteryield be executed after the response is sent.

Use Cases with Early Exit Code¶

There are some use cases with specific conditions that could benefit from the old behavior of running the exit code of dependencies withyield before sending the response.

For example, imagine you have code that uses a database session in a dependency withyield only to verify a user, but the database session is never used again in thepath operation function, only in the dependency,and the response takes a long time to be sent, like aStreamingResponse that sends data slowly, but for some reason doesn't use the database.

In this case, the database session would be held until the response is finished being sent, but if you don't use it, then it wouldn't be necessary to hold it.

Here's how it could look like:

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))

The exit code, the automatic closing of theSession in:

# 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))

...would be run after the response finishes sending the slow data:

# 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))

But asgenerate_stream() doesn't use the database session, it is not really necessary to keep the session open while sending the response.

If you have this specific use case using SQLModel (or SQLAlchemy), you could explicitly close the session after you don't need it anymore:

# 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))

That way the session would release the database connection, so other requests could use it.

If you have a different use case that needs to exit early from a dependency withyield, please create aGitHub Discussion Question with your specific use case and why you would benefit from having early closing for dependencies withyield.

If there are compelling use cases for early closing in dependencies withyield, I would consider adding a new way to opt in to early closing.

Dependencies withyield andexcept, Technical Details¶

Before FastAPI 0.110.0, if you used a dependency withyield, and then you captured an exception withexcept in that dependency, and you didn't raise the exception again, the exception would be automatically raised/forwarded to any exception handlers or the internal server error handler.

This was changed in version 0.110.0 to fix unhandled memory consumption from forwarded exceptions without a handler (internal server errors), and to make it consistent with the behavior of regular Python code.

Background Tasks and Dependencies withyield, Technical Details¶

Before FastAPI 0.106.0, raising exceptions afteryield was not possible, the exit code in dependencies withyield was executedafter the response was sent, soException Handlers would have already run.

This was designed this way mainly to allow using the same objects "yielded" by dependencies inside of background tasks, because the exit code would be executed after the background tasks were finished.

This was changed in FastAPI 0.106.0 with the intention to not hold resources while waiting for the response to travel through the network.

If you used to rely on this behavior, now you should create the resources for background tasks inside the background task itself, and use internally only data that doesn't depend on the resources of dependencies withyield.

For example, instead of using the same database session, you would create a new database session inside of the background task, and you would obtain the objects from the database using this new session. And then instead of passing the object from the database as a parameter to the background task function, you would pass the ID of that object and then obtain the object again inside the background task function.