AttributeError：“Depends”对象没有属性“query”FastAPI

Question

所以我尝试在这里编写简单的函数，但是每次我运行 swagger 时，都会出现上述错误。

这是我的功能：

def authenticate_user(username: str, password: str, db: Session = Depends(bd.get_db)):
    user = db.query(bd.User.username).filter(username == username).first()
    if not user:
        return False
    if not verify_password(password, user.password_hash):
        return False
    return user

这是我的 get_db 函数，它非常标准：

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

我注意到

Depends(bd.get_db)

在端点函数（带有

@app.post

/

@app.get

装饰器的函数）中工作得很好，但在普通函数中却不起作用。

显然，我不太理解依赖注入的概念，但我还不能完全掌握它。

Answer 1

这个页面对我帮助很大，https://github.com/tiangolo/fastapi/issues/1693#issuecomment-665833384

你不能在自己的函数中使用 Depends，它必须在 FastAPI 中 功能，主要是路线。但是，您可以在自己的中使用 Depends 当该函数也是一个依赖项时，可以有一个 功能链。

例如，路由使用 Depends 来解析“getcurrentuser”，这也 使用 Depends 来解析 'getdb'，整个链将被解析。 但是，如果您随后在不使用 Depends 的情况下调用“getcurrentuser”，则不会 能够解析“getdb”。

我所做的就是从路由中获取数据库会话，然后将其传递下去 贯穿每一层和功能。我相信这个也更好 设计。

Answer 2

我在尝试运行也被定义为端点的后台任务时遇到了类似的问题。为了规避这个问题，我所做的就是围绕它创建一个包装器。我不确定您面临什么情况，但我认为我的例子可以提供帮助。

这是我的代码：

用于依赖注入的数据库函数

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

没有端点装饰器的函数定义

async def function(args, db:Session):
    ... do stuff ...
    ... interact with db ...
    return "success!"

端点功能

@app.post("/function_endpoint/")
async def function_endpoint(args, db:Session=Depends(get_db)):
    return await function(args, db)

现在，我可以调用

function

，而不会丢失端点功能和重复代码

需要后台任务的复杂例程

@app.post("/complex_routine/")
async def complex_routine_endpoint(
    args, background_tasks:BackgroundTasks, db:Session=Depends(get_db)
):
    ... do complex stuff ...
    ... call other functions and pass db ...
    background_tasks.add_task(function, args, db)
    return "all good!"

Answer 3

fastapi.Depends

函数是FastAPI依赖注入系统的一部分。其记录的用例位于 FastAPI“端点”函数的签名中，即用

@app.get

等装饰的函数。FastAPI 依赖注入在没有该装饰器的函数中不起作用。

但是，如果您不需要依赖项注入，而只是想在任意函数中访问数据库，则可以为其提供一个有效的数据库会话实例，例如由您的

get_db

生成器返回的一个：

generator = get_db()
session = next(generator)
authenticate_user("bob", "***", session)

上面的代码应该与您原来的

authenticate_user

和

get_db

定义一起使用（尽管我自己没有测试它），但是您可以将前两个赋值移到函数体中，并从其函数中删除

db

参数签名。

Answer 4

from fastapi import Depends
from fastapi.params import Depends as DependsType
from sqlmodel import Session


class SessionMixin:
    __session: Session = Depends(get_session)

    @property
    def session(self) -> Session:
        if isinstance(self.__session, DependsType):
            return next(get_session())
        return self.__session