AttributeError: 'Depends' object has no attribute 'query' FastAPI

Question

So I am trying to write simple function here, but everytime I run swagger I got above mentioned error.

Here's my function:

def authenticate_user(username: str, password: str, db: Session = Depends(bd.get_db)):
    user = db.query(bd.User.username).filter(username == username).first()
    if not user:
        return False
    if not verify_password(password, user.password_hash):
        return False
    return user

and here's my get_db function it is pretty standard:

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

I've noticed that Depends(bd.get_db) works perfectly fine within endpoint functions (the ones with @app.post/@app.get decorators), but somehow doesn't work within plain functions. Apparently I don't quite understand the concept of dependency injections, but I can't quite grasp it yet.

Any help would be appreciated. Thank you kindly!

Answer 1

This page helped me a lot, https://github.com/tiangolo/fastapi/issues/1693#issuecomment-665833384

you can't use Depends in your own functions, it has to be in FastAPI functions, mainly routes. You can, however, use Depends in your own functions when that function is also a dependency, so could can have a chain of functions.

Eg, a route uses Depends to resolve a 'getcurrentuser', which also uses Depends to resolve 'getdb', and the whole chain will be resolved. But if you then call 'getcurrentuser' without using Depends, it won't be able to resolve 'getdb'.

What I do is get the DB session from the route and then pass it down through every layer and function. I believe this is also better design.

Answer 2

I had a similar problem trying to run a background task which was also defined as an endpoint. What I did to circumvent the issue was to create a wrapper around it. I'm not sure what scenario are you facing, but I think my example can help.

Here is my code:

DB Function for dependency injection

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

Function definition without endpoint decorator

async def function(args, db:Session):
    ... do stuff ...
    ... interact with db ...
    return "success!"

Endpoint function

@app.post("/function_endpoint/")
async def function_endpoint(args, db:Session=Depends(get_db)):
    return await function(args, db)

Now, I can call function without loosing endpoint functionality and duplicate code

Complex routine in need of background task

@app.post("/complex_routine/")
async def complex_routine_endpoint(
    args, background_tasks:BackgroundTasks, db:Session=Depends(get_db)
):
    ... do complex stuff ...
    ... call other functions and pass db ...
    background_tasks.add_task(function, args, db)
    return "all good!"