pythondependency-injectionsqlalchemyfastapi

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


I'm new to this.

Here is my generator function to retrieve database object:

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

Then I have a method to get a user from my database:

def get_user_from_db(email: str, db: Session = Depends(get_db_session)):
    user = db.query(User).filter(User.email == email).first()
    return user

Then I have my route, where I use dependency:

@app.post("/initialize_user")
def initialize_user(email: str, db: Session = Depends(get_db_session)):
    # Check if the user already exists in the database
    existing_user = get_user_from_db(email, db)

    if existing_user:
        # User already exists, return their information
        return {
            "email": existing_user.email,
            "credit_balance": existing_user.credit_balance,
            "subscription_type": existing_user.subscription_type
        }
    else:
        # Initialize a new user with default values
        new_user = User(email=email, credit_balance=20, subscription_type="trial")
        db.add(new_user)
        db.commit()
        db.refresh(new_user)
        return {
            "email": new_user.email,
            "credit_balance": new_user.credit_balance,
            "subscription_type": new_user.subscription_type
        }

And I received an error:

user = db.query(User).filter(User.email == email).first()
AttributeError: 'Depends' object has no attribute 'query'

According to this post, I can't use dependencies in my own functions, so I tried to use them in routes, like this:

@app.post("/initialize_user")
def initialize_user(email: str, db: Session = Depends(get_db_session)):
            # Check if the user already exists in the database
            existing_user = db.query(User).filter(User.email == email).first()
    ...
    }

And I received the same error.

What I'm doing wrong?


Solution

  • As @MatsLindh suggested in his comments to my original post, using TestClient solved my problem.

    Here is how for example the test of initializing the user, and getting credit balance:

    def test_update_and_get_credit_balance():
        # Initialize the user with a certain amount of credits
        response_initialize = client.post("/initialize_user?email=test@example.com")
        assert response_initialize.status_code == 200
    
        # Call update_credit_balance to deduct one credit
        response_update = client.post("/update_credit_balance?email=test@example.com")
        assert response_update.status_code == 200
    
        # Call get_credit_balance to retrieve the current balance
        response_get_balance = client.get("/get_credit_balance?email=test@example.com")
        assert response_get_balance.status_code == 200
    
        # Print the response content
        print("Response Content - test user's credit balance:")
        print(response_get_balance.text)
    

    And then in the same script how you can run your tests:

    # Run the tests
    if __name__ == "__main__":
        print('I am in main')
        test_initialize_user()
        test_get_credit_balance()
        test_update_and_get_credit_balance()
        test_convert_to_premium()
        test_update_and_get_credit_balance()