FastAPI使用OAuth2 Scope报错：'Depends'对象无'query'属性

1. AttributeError: 'Depends' object has no attribute 'query' 错误修复 #

这个错误的核心是代码中错误地将Depends对象本身当作数据库会话/用户对象调用query方法，而非使用Depends注入的实际实例。结合/bike/check正常、/bike/add出错的情况，大概率是/bike/add的依赖注入写法有误。

常见错误场景及修正： #

• 错误写法示例：

  @app.post("/bike/add")
  def add_bike(db=Depends(get_db), current_user=Depends(get_current_user)):
      # 此处db是Depends对象，而非Session实例，调用query会报错
      bike = db.query(Bike).filter(...)
  

• 正确写法：

  from sqlalchemy.orm import Session
  
  @app.post("/bike/add")
  def add_bike(db: Session = Depends(get_db), current_user = Depends(get_current_user)):
      # 加上Session类型注解后，FastAPI会自动注入真实的数据库会话
      bike = db.query(Bike).filter(...)
  

同时检查get_user方法实现，确保返回的是用户模型实例，而非Depends相关对象：

def get_user(db: Session = Depends(get_db), username: str = Depends(oauth2_scheme)):
    user = db.query(User).filter(User.username == username).first()
    if not user:
        raise HTTPException(status_code=401, detail="无效用户")
    return user

2. 异步接口协程错误修复 #

将接口改为异步后，需确保数据库操作适配异步模式：

• 使用SQLAlchemy异步引擎时，必须用AsyncSession替代同步Session，且查询需加await：

  from sqlalchemy.ext.asyncio import AsyncSession
  
  # 异步数据库会话依赖
  async def get_async_db():
      async with AsyncSession(engine) as session:
          yield session
  
  # 异步get_user实现
  async def get_user(db: AsyncSession = Depends(get_async_db), username: str = Depends(oauth2_scheme)):
      user = await db.query(User).filter(User.username == username).first()
      if not user:
          raise HTTPException(status_code=401, detail="无效用户")
      return user
  
  # 异步接口写法
  @app.post("/bike/add")
  async def add_bike(db: AsyncSession = Depends(get_async_db), current_user = Depends(get_current_user)):
      new_bike = Bike(name="xxx")
      db.add(new_bike)
      await db.commit()
      await db.refresh(new_bike)
      return new_bike
  

• 若必须保留同步数据库操作，需用asyncio.to_thread包装避免协程阻塞：

  import asyncio
  
  @app.post("/bike/add")
  async def add_bike(db: Session = Depends(get_db), current_user = Depends(get_current_user)):
      def sync_add_bike():
          new_bike = Bike(name="xxx")
          db.add(new_bike)
          db.commit()
          db.refresh(new_bike)
          return new_bike
      
      bike = await asyncio.to_thread(sync_add_bike)
      return bike
  

3. OAuth2 Scope 验证逻辑检查 #

确保/bike/add的Scope配置与验证逻辑正确：

from fastapi import Security
from fastapi.security import OAuth2PasswordBearer, SecurityScopes

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"bike:add": "允许添加自行车", "bike:check": "允许检查自行车"}
)

async def get_current_user(
    security_scopes: SecurityScopes,
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_async_db)
):
    authenticate_value = f'Bearer scope="{security_scopes.scope_str}"' if security_scopes.scopes else "Bearer"
    # 解析token、获取用户逻辑...
    user = await get_user(db, username=token_data.username)
    # 校验用户权限
    for scope in security_scopes.scopes:
        if scope not in user.scopes:
            raise HTTPException(
                status_code=403,
                detail="权限不足",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user

# 为/add接口指定所需Scope
@app.post("/bike/add")
async def add_bike(
    current_user = Security(get_current_user, scopes=["bike:add"]),
    db: AsyncSession = Depends(get_async_db)
):
    # 接口业务逻辑

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