如何异步使用FastAPI Depends链？

Question

如标题所示，我想使用FastAPI的

Depends

来实现针对特定路由的基于角色的认证控制。

我有两个文件。如果我写的认证逻辑如下所示：

auth.py 文件

 class Authenticator:   
    async def get_sub(self, token: Annotated[str, Depends(oauth2_scheme)]) -> str:
        ** asynchronous operation **
        return sub

    async def get_user(self, sub: Annotated[User, Depends(get_sub)]) -> UserInDB:
        sub = await self.get_sub(token)
        user = await self.user_db.get_user_by_id(id=id)
        if not user:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not register yet")
        return user
authenticator = Authenticator()

router.py 文件

router = APIRouter(prefix="/user", dependencies=[Depends(authenticator.get_user)])



@router.get("/")
async def protected_route(request: Request):
    pass

如果我在第一个文件中使用

get_user

编写

sub: Annotated[User, Depends(get_sub)]

函数，那么当我运行它时，路径中会出现一个“self”查询参数，这看起来很奇怪。如何使这个异步

Depends

链正常工作而不出现“自我”问题？

了解异步

Depends

链如何工作

Answer 1

如果不需要在

self

中使用

get_sub

，可以将

get_sub

设为静态方法：

class Authenticator:
    @staticmethod
    async def get_sub(token: Annotated[str, Depends(oauth2_scheme)]) -> str:
        ** asynchronous operation **
        return sub

    async def get_user(self, sub: Annotated[User, Depends(get_sub)]) -> UserInDB:
        sub = await self.get_sub(token)
        user = await self.user_db.get_user_by_id(id=id)
        if not user:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not register yet")
        return user
authenticator = Authenticator()

另一种方法是在没有类的情况下实现它：

db = UserDatabase()


async def get_user_db():
    yield db

async def get_sub(token: Annotated[str, Depends(oauth2_scheme)]) -> str:
    #
    return sub

async def get_user(
    sub: Annotated[User, Depends(get_sub)],
    user_db: Annotated[UserDatabase, Depends(get_user_db)],
) -> UserInDB:
    user = await user.get_user_by_id(id=id)
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not register yet")
    return user
    

app = FastAPI()

@app.get("/")
def index(user: Annotated[UserInDB, Depends(get_user)]):
    pass