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