2.14. FastAPI Dependencies

  • CDI - Content Dependency Injection

  • Used when you have functions which takes the same thing (for example parameters)

  • If one of your dependencies is declared multiple times for the same path operation, for example, multiple dependencies have a common sub-dependency, FastAPI will know to call that sub-dependency only once per request.

  • And it will save the returned value in a "cache" and pass it to all the "dependants" that need it in that specific request, instead of calling the dependency multiple times for the same request.

2.14.1. Function

>>> from fastapi import Depends, FastAPI
>>>
>>> app = FastAPI()
>>>
>>>
>>> async def common_parameters(q: str | None = None,
...                             skip: int = 0,
...                             limit: int = 100) -> dict:
...     return {"q": q, "skip": skip, "limit": limit}
>>>
>>>
>>> @app.get("/items/")
... async def read_items(commons: dict = Depends(common_parameters)):
...     return commons
>>>
>>>
>>> @app.get("/users/")
... async def read_users(commons: dict = Depends(common_parameters)):
...     return commons

2.14.2. Class

>>> from fastapi import Depends, FastAPI
>>>
>>> app = FastAPI()
>>>
>>>
>>> fake_items_db = [
...     {"item_name": "Foo"},
...     {"item_name": "Bar"},
...     {"item_name": "Baz"}]
>>>
>>>
>>> class CommonQueryParams:
...     def __init__(self, q: str | None = None,
...                  skip: int = 0,
...                  limit: int = 100):
...         self.q = q
...         self.skip = skip
...         self.limit = limit
>>>
>>>
>>> @app.get("/items/")
... async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
...     response = {}
...     if commons.q:
...         response.update({"q": commons.q})
...     items = fake_items_db[commons.skip : commons.skip + commons.limit]
...     response.update({"items": items})
...     return response

2.14.3. Shortcut

  • FastAPI provides a shortcut injecting classes

  • commons: CommonQueryParams = Depends()

Instead of writing:

>>> commons: CommonQueryParams = Depends(CommonQueryParams)

you write:

>>> commons: CommonQueryParams = Depends()

2.14.4. Dependencies in path operation decorators

  • In some cases you don't really need the return value of a dependency inside your path operation function.

  • Or the dependency doesn't return a value.

  • But you still need it to be executed/solved.

  • For those cases, instead of declaring a path operation function parameter with Depends, you can add a list of dependencies to the path operation decorator.

>>> from fastapi import Depends, FastAPI, Header, HTTPException
>>>
>>> app = FastAPI()
>>>
>>>
>>> async def verify_token(x_token: str = Header(...)):
...     if x_token != "fake-super-secret-token":
...         raise HTTPException(status_code=400, detail="X-Token header invalid")
>>>
>>>
>>> async def verify_key(x_key: str = Header(...)):
...     if x_key != "fake-super-secret-key":
...         raise HTTPException(status_code=400, detail="X-Key header invalid")
...     return x_key
>>>
>>>
>>> @app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
... async def read_items():
...     return [{"item": "Foo"}, {"item": "Bar"}]

2.14.5. Global Dependencies

  • For some types of applications you might want to add dependencies to the whole application.

>>> from fastapi import Depends, FastAPI, Header, HTTPException
>>>
>>>
>>> async def verify_token(x_token: str = Header(...)):
...     if x_token != "fake-super-secret-token":
...         raise HTTPException(status_code=400, detail="X-Token header invalid")
>>>
>>>
>>> async def verify_key(x_key: str = Header(...)):
...     if x_key != "fake-super-secret-key":
...         raise HTTPException(status_code=400, detail="X-Key header invalid")
...     return x_key
>>>
>>>
>>> app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
>>>
>>>
>>> @app.get("/items/")
... async def read_items():
...     return [{"item": "Portal Gun"}, {"item": "Plumbus"}]
>>>
>>>
>>> @app.get("/users/")
... async def read_users():
...     return [{"username": "Rick"}, {"username": "Morty"}]

2.14.6. Router Based Dependencies

>>> from fastapi import APIRouter, Depends, HTTPException
>>> from ..dependencies import get_token_header
>>>
>>> router = APIRouter(
...     prefix="/items",
...     tags=["items"],
...     dependencies=[Depends(get_token_header)],
...     responses={404: {"description": "Not found"}},
... )
>>>
>>>
>>> fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}
>>>
>>>
>>> @router.get("/")
... async def read_items():
...     return fake_items_db
>>>
>>>
>>> @router.get("/{item_id}")
... async def read_item(item_id: str):
...     if item_id not in fake_items_db:
...         raise HTTPException(status_code=404, detail="Item not found")
...     return {"name": fake_items_db[item_id]["name"], "item_id": item_id}
>>>
>>>
>>> @router.put(
...     "/{item_id}",
...     tags=["custom"],
...     responses={403: {"description": "Operation forbidden"}},
... )
... async def update_item(item_id: str):
...     if item_id != "plumbus":
...         raise HTTPException(
...             status_code=403, detail="You can only update the item: plumbus"
...         )
...     return {"item_id": item_id, "name": "The great Plumbus"}
>>> from fastapi import Depends, FastAPI
>>> from .dependencies import get_query_token, get_token_header
>>> from .internal import admin
>>> from .routers import items, users
>>>
>>> app = FastAPI(dependencies=[Depends(get_query_token)])
>>>
>>>
>>> app.include_router(users.router)
>>> app.include_router(items.router)
>>> app.include_router(
...     admin.router,
...     prefix="/admin",
...     tags=["admin"],
...     dependencies=[Depends(get_token_header)],
...     responses={418: {"description": "I'm a teapot"}},
... )
>>>
>>>
>>> @app.get("/")
... async def root():
...     return {"message": "Hello Bigger Applications!"}

2.14.7. Dependencies with yield

  • FastAPI supports dependencies that do some extra steps after finishing.*

  • To do this, use yield instead of return, and write the extra steps after.

  • It might be tempting to raise an HTTPException or similar in the exit code, after the yield. But it won't work.

  • The exit code in dependencies with yield is executed after the response is sent

  • Only one response will be sent to the client.

  • After one of those responses is sent, no other response can be sent.

>>> async def get_db():
...     db = DBSession()
...     try:
...         yield db
...     finally:
...         db.close()
>>> class MyDatabase:
...     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 MyDatabase() as db:
...         yield db