FASTApi认证注入

Question

我的应用程序有一个 AuthenticateService 实现如下：

from domain.ports.repositories import ISalesmanRepository
from fastapi import HTTPException, status
from fastapi_jwt_auth import AuthJWT
from fastapi_jwt_auth.exceptions import JWTDecodeError
from shared.exceptions import EntityNotFound
from adapters.api.authentication.config import User


class AuthenticateService:
    def __init__(self, user_repo: ISalesmanRepository):
        self._repo = user_repo

    def __call__(self, auth: AuthJWT) -> User:
        credentials_exception = HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
        try:
            auth.jwt_required()
            user_id = auth.get_jwt_subject()
        except JWTDecodeError:
            raise credentials_exception

        try:
            user = self._repo.get_by_id(user_id)
            return user
        except EntityNotFound:
            raise credentials_exception

所以行为基本上是：

jwt 是否有效，从存储库获取用户并 returns
如果 jwt 无效则抛出 401

问题是在每个实现的控制器中我都必须重复这个过程。我试图实现一个装饰器，在成功的情况下将用户注入控制器，但我做不到。我确信实现它的最佳方法是使用 fastAPI 的 Depends 依赖注入器。今天，控制器看起来像这样：

from typing import Optional

from adapters.api.services import authenticate_service, create_sale_service
from fastapi import APIRouter, Depends
from fastapi_jwt_auth import AuthJWT
from pydantic import BaseModel

router = APIRouter()


class Request(BaseModel):
    code: str
    value: float
    date: str
    status: Optional[str] = None


@router.post('/sale')
def create_sale(request: Request, auth: AuthJWT = Depends()):
    user = authenticate_service(auth)
    result = create_sale_service.handle(
        {"salesman": user, "sale": request.dict()}
    )
    return result.dict()

如何抽象我的身份验证，使我的控制器看起来像以下任何版本：


# Option 1: decorator

@router.post('/sale')
@authentication_required
def create_sale(request: Request, user: User): # User is the `__call__` response from `AuthenticateService` class
    result = create_sale_service.handle(
        {"salesman": user, "sale": request.dict()}
    )
    return result.dict()


# Option 2:
@router.post('/sale')
def create_sale(request: Request, user: User = Depends(authenticate_service)): # Something like that, using the depends to inject User to me
    result = create_sale_service.handle(
        {"salesman": user, "sale": request.dict()}
    )
    return result.dict()

Answer 1

所以我多读了一些 Depends 文档，意识到我尝试在控制器签名上注入用户的尝试出了什么问题。

正确的实施方式：

> AuthService class

class AuthenticateService:
    def __init__(self, user_repo: ISalesmanRepository):
        self._repo = user_repo

    def __call__(self, auth: AuthJWT = Depends()) -> User:
    ...

authenticate_service = AuthenticateService(user_repository)

> Controller

@router.post('/sale')
def create_sale(request: Request, user: User = Depends(authenticate_service)):
    result = create_sale_service.handle(
        {"salesman": user, "sale": request.dict()}
    )
    return result.dict()

FASTApi认证注入

FASTApi authentication injection

python

fastapi