python - FastAPI - 编写具有多个条件的 REST API 的最佳实践

Question

假设我有两个实体，Users 和 Councils，以及一个 M2M 关联表 UserCouncils。 Users 可以从 Councils 添加/删除，只有管理员可以这样做(在 UserCouncil 的 role 属性中定义> 关系)。 现在，在为 /councils/{council_id}/remove 创建端点时，我面临着在操作前检查多个约束的问题，例如:

@router.delete("/{council_id}/remove", response_model=responses.CouncilDetail)
def remove_user_from_council(
    council_id: int | UUID = Path(...),
    *,
    user_in: schemas.CouncilUser,
    db: Session = Depends(get_db),
    current_user: Users = Depends(get_current_user),
    council: Councils = Depends(council_id_dep),
) -> dict[str, Any]:
    """

    DELETE /councils/:id/remove (auth)

    remove user with `user_in` from council
    current user must be ADMIN of council
    """

    # check if input user exists
    if not Users.get(db=db, id=user_in.user_id):
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
        )

    if not UserCouncil.get(db=db, user_id=user_in.user_id, council_id=council.id):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Cannot delete user who is not part of council",
        )

    # check if current user exists in council
    if not (
        relation := UserCouncil.get(
            db=db, user_id=current_user.id, council_id=council.id
        )
    ):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Current user not part of council",
        )

    # check if current user is Admin
    if relation.role != Roles.ADMIN:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="Unauthorized"
        )

    elif current_user.id == user_in.user_id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Admin cannot delete themselves",
        )

    else:
        updated_users = council.remove_member(db=db, user_id=user_in.user_id)
        result = {"council": council, "users": updated_users}
        return result

这些检查是不言自明的。但是，这会在端点定义中添加大量代码。端点定义通常应该是简约的吗？我可以将所有这些检查包装在 Councils crud 方法(即 council.remove_member())中，但这意味着在里面添加 HTTPException crud 类，我不想这样做。

解决此类情况的一般最佳做法是什么？我在哪里可以阅读更多相关信息？任何形式的帮助将不胜感激。

谢谢。

Answer 1

那么，我将通过您的示例告诉您我将如何去做。

一般来说，我喜欢将端点保持在最低限度。您要使用的是构建 API 时使用的常见模式，即将您的业务逻辑捆绑到一个服务类中。该服务类允许您重用逻辑。假设您想从队列或 cron 作业中删除委员会成员。这会引出您强调的下一个问题，即在您的服务类中存在 HTTP 特定异常，这些异常可能不会在 HTTP 上下文中使用。幸运的是，这不是一个很难解决的问题，您可以定义自己的异常并要求 API 框架捕获它们，以便重新引发所需的 HTTP 异常。

定义自定义异常:

class UnauthorizedException(Exception):
    def __init__(self, message: str):
        super().__init__(message)
        self.message = message


class InvalidActionException(Exception):
    ...


class NotFoundException(Exception):
    ...

在 Fast API 中，您可以捕获应用程序抛出的特定异常

@app.exception_handler(UnauthorizedException)
async def unauthorized_exception_handler(request: Request, exc: UnauthorizedException):
    return JSONResponse(
            status_code=status.HTTP_403_FORBIDDEN,
            content={"message": exc.message},
    )

@app.exception_handler(InvalidActionException)
async def unauthorized_exception_handler(request: Request, exc: InvalidActionException):
    ...

使用合理的方法将您的业务逻辑包装到服务类中，并引发您为服务定义的异常

class CouncilService:
    def __init__(self, db: Session):
        self.db = db

    def ensure_admin_council_member(self, user_id: int, council_id: int):
        # check if current user exists in council
        if not (
                relation := UserCouncil.get(
                        db=self.db, user_id=user_id, council_id=council_id
                )
        ):
            raise UnauthorizedException("Current user not part of council")

        # check if current user is Admin
        if relation.role != Roles.ADMIN:
            raise UnauthorizedException("Unauthorized")

    def remove_council_member(self, user_in: schemas.CouncilUser, council: Councils):
        # check if input user exists
        if not Users.get(db=self.db, id=user_in.user_id):
            raise NotFoundException("User not found")

        if not UserCouncil.get(db=self.db, user_id=user_in.user_id, council_id=council.id):
            raise InvalidActionException("Cannot delete user who is not part of council")

        if current_user.id == user_in.user_id:
            raise InvalidActionException("Admin cannot delete themselves")

        updated_users = council.remove_member(db=self.db, user_id=user_in.user_id)
        result = {"council": council, "users": updated_users}
        return result

最后你的端点定义非常精简

编辑:从路径中删除了 /remove 动词，正如评论中指出的那样，已经指定了动词。理想情况下，您的路径应包含引用资源的名词。

@router.delete("/{council_id}", response_model=responses.CouncilDetail)
def remove_user_from_council(
    council_id: int | UUID = Path(...),
    *,
    user_in: schemas.CouncilUser,
    current_user: Users = Depends(get_current_user),
    council: Councils = Depends(council_id_dep),
    council_service: CouncilService = Depends(get_council_service),
) -> responses.CouncilDetail:
    """

    DELETE /councils/:id (auth)

    remove user with `user_in` from council
    current user must be ADMIN of council
    """
    council_service.ensure_admin_council_member(current_user.id, council_id)
    return council_service.remove_council_member(user_in, council)