取决于 FastAPI 中的 class 个实例

Question

我正在努力为 return 具有 class 实施的经过验证的用户找到一个体面的实施。以前它使用简单的功能，但我想重构这一部分。所以我添加了一个 class 并试图让它工作。问题是它似乎没有创建 class 的新实例。如果我正在尝试做 Depends(UserAuthService().test) 我收到了一个错误

{
    "message": "Bad request.",
    "details": "'TokenService' object is not callable"
}

router.py

router = APIRouter()


@router.get("/verify")
def verify_user(user: User = Depends(UserAuthService().get_current_valid_user)):
    return user

user_auth_service.py

class UserAuthService:
    def __init__(self):
        self.user_repository = UserRepository()
        self.token_service = TokenService()

    def get_current_user(self, token: str = Depends(OAUTH2_SCHEME)):
        credentials_exception = HTTPException(
            status_code=HTTP_403_FORBIDDEN, detail="Could not validate the token"
        )
        user_does_not_exist_exception = HTTPException(
            status_code=HTTP_403_FORBIDDEN, detail="User does not exist"
        )
        try:
            payload = self.token_service.decode_access_token(token)
            sub: int = payload.get("sub")
            if sub is None:
                raise credentials_exception
        except JWTError:
            raise credentials_exception

        user = self.user_repository.get_user(sub)
        if user is None:
            raise user_does_not_exist_exception
        return User(**user)

    def get_current_valid_user(self, user: User = Depends(get_current_user)):
        if user.disabled:
            raise HTTPException(status_code=400, detail="Inactive user")
        return user

    def test(self):
        return self.token_service('123456')

P.S。如您所见，class 方法中有 Depends，但那是来自以前的功能实现。

Answer 1

问题是：

class UserAuthService:
    def __init__(self):
        ...
        self.token_service = TokenService()

    ...

    def test(self):
        return self.token_service('123456')

您创建实例
您尝试调用实例而不是它的可调用方法

这里有很多解决方案，这取决于你想做什么：

运行正确的实例方法而不是构造函数：

class A():

  def __init__(self, param = None):
     self.param = param

  def print_something(self, something = None):
     print('Init param is', self.param)
     print('Something is', something)


class B():

  def __init__(self, param = None):
    self.a_instance = A(param)

  def test(self, something = None):
    self.a_instance.print_something(something) # run a_instance.print_something() instead of a_instance()

b = B('A obj')
b.test('test')

Init param is A obj
Something is test

设置可调用实例：

class ACallable():

  def __init__(self, param = None):
     self.param = param

  def __call__(self, something = None):
     print('Init param is', self.param)
     print('Something is', something)


class BCalling():

  def __init__(self, param = None):
    self.a_instance = ACallable(param)

  def test(self, something = None):
    self.a_instance(something)  # __call__ method will be called

b = BCalling('A obj')
b.test('test')

Init param is A obj
Something is test

将class设置为构造函数参数而不是调用subclass的构造函数并创建对象：

class A():

  def __init__(self, param = None):
     print('Init param is', param)

class B():

  def __init__(self, param = None):
    self.a_instance = A  # here is A without brackets

  def test(self, param = None):
    self.a_instance(param)  # here is a_instance (== A) with brackets

b = B()
b.test('<- called from constructor')

Init param is <- called from constructor

Answer 2

修复后类最终代码如下所示

router.py

@router.get("/verify")
def verify_user(user: User = Depends(UserValidator())):
    return user

user_auth_service.py 和 user_validator.py

class UserAuthService:
    def __init__(self):
        self.user_repository = UserRepository()
        self.token_service = TokenService()
        self.oauth2_scheme = OAuth2BearerCookie()

    async def get_current_user(self, request: Request):
        token = await self.oauth2_scheme(request)
        credentials_exception = HTTPException(
            status_code=HTTP_403_FORBIDDEN, detail="Could not validate the token"
        )
        user_does_not_exist_exception = HTTPException(
            status_code=HTTP_403_FORBIDDEN, detail="User does not exist"
        )
        try:
            payload = self.token_service.decode_access_token(token)
            sub: int = payload.get("sub")
            if sub is None:
                raise credentials_exception
        except JWTError:
            raise credentials_exception

        user = self.user_repository.get_user(sub)
        if user is None:
            raise user_does_not_exist_exception
        return User(**user)

    async def get_current_valid_user(self, request: Request):
        user = await self.get_current_user(request)
        if user.disabled:
            raise HTTPException(status_code=400, detail="Inactive user")
        return user


class UserValidator:
    async def __call__(self, request: Request):
        user_service = UserAuthService()
        return await user_service.get_current_valid_user(request)

取决于 FastAPI 中的 class 个实例

Depends and a class instance in FastAPI

python

fastapi

问题是：

这里有很多解决方案，这取决于你想做什么：