FastAPI - Best practices for writing REST APIs with multiple conditions-CodePudding

Let's say I have two entities, Users and Councils, and a M2M association table UserCouncils. Users can be added/removed from Councils and only admins can do that (defined in a role attribute in the UserCouncil relation). Now, when creating endpoints for /councils/{council_id}/remove, I am faced with the issue of checking multiple constraints before the operation, such as the following:


@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

These checks are pretty self-explanatory. However, this adds a lot of code in the endpoint definition. Should the endpoint definitions be generally minimalistic? I could wrap all these checks inside the Councils crud method (i.e., council.remove_member()), but that would mean adding HTTPExceptions inside crud classes, which I don't want to do.

What are the general best practices for solving situations like these, and where can I read more about this? Any kind of help would be appreciated.

Thanks.

CodePudding user response：

So, I will tell you how I would go about doing it with your example.

Generally I like to keep my endpoints quite minimal. What you what to employ is a common pattern used in building APIs and that is to bundle your business logic into a service class. This service class allows you to reuse logic. Say you want to remove a council member from a queue or cron job. This brings up the next issue you highlighted and that is about having HTTP specific exceptions in your service class which may not be used in an HTTP context. Fortunately this is not a difficult one to solve, you can just define your own exceptions and ask the API framework to catch them only to re-raise the desired HTTP exception.

Define a custom exception:

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


class InvalidActionException(Exception):
    ...


class NotFoundException(Exception):
    ...

In Fast API you can catch specific exceptions your application throws

@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):
    ...

Wrap up your business logic into a service class with sensible methods and raise the exceptions you have defined for your service

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

and then finally your endpoint definition is quite lean

EDIT: removed the /remove verb from the path, as pointed out in the comments, the verb is already specified. Ideally your path should contain Nouns referring to the resource.

@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)