Dependencies Injection

Simple Dependency

Here is an example of declaring a function that will then be used to pass the same parameters to other route functions:

from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    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

Dependency in class

Here is the same example, but using a 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

The e expression that the dependency is injected could be simplified as follows:

async def read_items(commons: CommonQueryParams = Depends()):

Nested dependencies

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()

def query_extractor(q: str | None = None):
    return q

def query_or_cookie_extractor(
    q: str = Depends(query_extractor), last_query: str | None = Cookie(default=None)
):
    if not q:
        return last_query
    return q

@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
    return {"q_or_cookie": query_or_default}

Dependencies on the route decorator

In cases where we need to execute the dependency but not use any return value from it, we can execute the dependency in the decorator of the route function:

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"}]

Dependency Global Enforcement

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

Dependencies with yield

In cases where you want to execute a dependency and after obtaining the response from the function, other functions are executed, in these cases yield is used:

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

Example to understand the execution flow:

from fastapi import Depends, FastAPI

async def verify_yield():
    print('1->First dependency lines')
    try:
        yield print('2->Yield dependency lines')
    finally:
        print('4->Finally lines')

app = FastAPI(dependencies=[Depends(verify_yield)])

@app.get("/users/")
async def read_users():
    print('3->Function lines')
    return [{"username": "Rick"}, {"username": "Morty"}]

Thanks for reading :)
I invite you to continue reading other entries and visiting us again soon.