FASTAPI OAuth认证中调用get_user时Depends对象无query属性报错
解决FastAPI中authenticate_user调用get_user时的数据库会话注入问题
错误原因
你直接把Depends(get_db)作为参数传给get_user,但Depends是FastAPI的依赖注入标记对象,不是实际的数据库Session实例,所以调用db.query()时会报错'Depends' object has no attribute 'query'。Depends只能在路由函数或者被FastAPI管理的依赖函数中自动解析,普通函数直接传它没用。
两种可行解决方案
方案一:让authenticate_user也接受依赖注入的数据库会话
修改authenticate_user,让它把db作为参数,然后在调用它的路由里通过Depends(get_db)传入实际的Session:
def authenticate_user(username: str, password: str, db: Session = Depends(get_db)): # 直接传入已经解析好的db会话 user = get_user(username, db) if not user: return False if not verify_password(password, user.password): return False return user
然后在你的登录路由里这样调用:
from fastapi import Depends, OAuth2PasswordRequestForm @app.post("/token") async def login_for_access_token( form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db) ): user = authenticate_user(form_data.username, form_data.password, db) if not user: # 处理认证失败逻辑,比如抛出HTTPException ... # 生成token等后续操作
方案二:在authenticate_user内部手动获取数据库会话
如果不想修改authenticate_user的参数,可以手动调用get_db生成器获取Session,用完后手动关闭:
def authenticate_user(username: str, password: str): # 手动获取数据库会话 db_generator = get_db() db = next(db_generator) try: user = get_user(username, db) if not user: return False if not verify_password(password, user.password): return False return user finally: # 手动关闭会话,避免连接泄漏 db.close()
额外优化建议
你的get_user函数里捕获SQLAlchemyError后返回了一个字典{"error": ...},这会导致authenticate_user里的if not user判断失效(因为字典是真值)。建议修改get_user的异常处理逻辑:
def get_user(nombre_usuario: str, db: Session): try: usuario_db = db.query(models.Usuarios).filter(models.Usuarios.nombre_usuario == nombre_usuario).first() return usuario_db # 不存在就返回None,让调用方处理 except SQLAlchemyError as e: print(f"数据库查询错误:{str(e)}") return None
这样authenticate_user里的if not user就能正确判断用户是否存在或查询出错的情况。
内容的提问来源于stack exchange,提问作者Jonatan Green
相关产品推荐
相关产品推荐

