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

那么，我将告诉你我将如何用你的例子来做.

一般来说，我喜欢保持我的端点非常小.您要使用的是构建API时使用的常见模式，即将业务逻辑绑定到服务类中.这个服务类允许您重用逻辑.假设您想从队列或cron作业(job)中删除一名委员会成员.这带来了您强调的下一个问题，即在您的服务类中存在可能不在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)