在pytest中调用FastAPI数据库报错：Depends对象无query属性

测试过程中尝试获取创建的测试用户以便在测试结束时删除，遇到错误：

FAILED tests_main.py::test_delete_new_users - AttributeError: 'Depends' object has no attribute 'query'

测试代码如下：

def test_delete_new_users(db: Session = Depends(database.get_db)):
    auth = client.post('/token',
                       data={'username': 'test123', 'password': 'test123'}
                       )
    access_token = auth.json().get('access_token')
    user = db.query(models.User).filter(models.User.username == 'test123').first()

    response = client.post('/user/delete/' + user.user_id,
                           headers={
                               'Authorization': 'bearer' + access_token
                           })
    assert response.status_code == 200

疑问：是否需要编写单独的API（比如get_user_id_with_username）通过用户名获取用户？或者有没有办法在pytest中直接进行数据库调用？因为了解到Depends仅可用于API调用。

• 直接在pytest中获取数据库会话，不要用Depends
  Depends是FastAPI用于接口层依赖注入的工具，不能直接用于pytest测试函数的参数注入。你可以在测试里直接调用database.get_db()生成会话实例：

  def test_delete_new_users():
      db = next(database.get_db())  # 从生成器中取出数据库会话
      auth = client.post('/token',
                         data={'username': 'test123', 'password': 'test123'}
                         )
      access_token = auth.json().get('access_token')
      user = db.query(models.User).filter(models.User.username == 'test123').first()
  
      # 注意Bearer后加空格，用f-string拼接更清晰
      response = client.post(f'/user/delete/{user.user_id}',
                             headers={
                                 'Authorization': f'Bearer {access_token}'
                             })
      assert response.status_code == 200
  

  用完会话后，可手动调用db.close()关闭，或者依赖数据库连接池自动管理。

• 无需额外编写API
  直接在测试中操作数据库是更高效的方案，避免为测试新增不必要的API接口，减少代码维护成本。

• 进阶：用pytest fixture管理数据库会话
  可以封装一个fixture来统一管理测试用的数据库会话，还能在测试前后自动清理数据，避免测试数据污染：

  import pytest
  from sqlalchemy.orm import Session
  from database import get_db
  
  @pytest.fixture(scope="function")
  def db_session():
      db = next(get_db())
      try:
          yield db
      finally:
          # 测试结束后回滚事务，清理测试数据
          db.rollback()
          db.close()
  
  def test_delete_new_users(db_session: Session):
      auth = client.post('/token',
                         data={'username': 'test123', 'password': 'test123'}
                         )
      access_token = auth.json().get('access_token')
      user = db_session.query(models.User).filter(models.User.username == 'test123').first()
  
      response = client.post(f'/user/delete/{user.user_id}',
                             headers={
                                 'Authorization': f'Bearer {access_token}'
                             })
      assert response.status_code == 200
  

内容的提问来源于stack exchange，提问作者Vedo