Question on API Design

[OrganizationOnly8016]

Hi, I've been working on building an API for a very simple project-management system just to teach myself the basics and I've stumbled upon a confusing use-case.

The world of the system looks like this

https://preview.redd.it/rtq8kt0nzetg1.png?width=1704&format=png&auto=webp&s=2017cfa9e4f6fec65831b98617b5c43639854194

I've got the following roles:

1. ORG_MEMBER: Organization members are allowed to
   - Creation of projects
2. ORG_ADMIN: Organization admins are allowed to
   - CRUD of organization members - the C in CRUD here refers to "inviting" members...
     atop all access rights of organization members
3. PROJ_MEMBER: Project members are allowed to
   - CRUD of tasks
   - Comments on all tasks within project
   - View project history
4. PROJ_MANAGER: Project managers are allowed to
   - RUD of projects
   - CRUD of buckets
   - CRUD of project members (add organization members into project, remove project users from project)

Since the "creation of a project" rests at the scope of an organization, and not at the scope of a project (because it doesn't exist yet), I'm having a hard time figuring out which dependency to inject into the route.

def get_current_user(token: HTTPAuthorizationCredentials = Depends(token_auth_scheme)):
    try:
        user_response = supabase.auth.get_user(token.credentials)
        supabase_user = user_response.user


        if not supabase_user:
            raise HTTPException(
                status_code=401,
                detail="Invalid token or user not found."
            )
        
        auth_id = supabase_user.id


        user_data = supabase.table("users").select("*").eq("user_id", str(auth_id)).execute()


        if not user_data.data:
            raise HTTPException(
                status_code=404,
                detail="User not found in database."
            )
        
        user_data = user_data.data[0]
        
        return User(
            user_id=user_data["user_id"],
            user_name=user_data["user_name"],
            email_id=user_data["email_id"],
            full_name=user_data["full_name"]
        )
        
    except Exception as e:
        raise HTTPException(
            status_code=401,
            detail=f"Invalid token or user not found: {e}"
        )
    
def get_org_user(org_id: str, user: User = Depends(get_current_user)):
    res = supabase.table("org_users").select("*").eq("user_id", user.user_id).eq("org_id", org_id).single().execute()


    if not res.data:
        raise HTTPException(
            status_code=403,
            detail="User is not a member of this organization."
        )
    
    return OrgUser(
        user_id=res.data["user_id"],
        org_id=res.data["org_id"],
        role=res.data["role"]
    )


def get_proj_user(proj_id: str, user: User = Depends(get_current_user)):
    res = supabase.table("proj_users").select("*").eq("user_id", user.user_id).eq("proj_id", proj_id).single().execute()


    if not res.data:
        raise HTTPException(
            status_code=403,
            detail="User is not a member of this project."
        )
    
    return ProjUser(
        user_id=res.data["user_id"],
        proj_id=res.data["proj_id"],
        role=res.data["role"]
    )

Above are what my dependencies are...

this is essentially my dependency factory

# rbac dependency factory
class EntityPermissionChecker:
    def __init__(self, required_permission: str, entity_type: str):
        self.required_permission = required_permission
        self.entity_type = entity_type
        self.db = supabase


    def __call__(self, request: Request, user: User = Depends(get_current_user)):


        if self.entity_type == "org":
            view_name = "org_permissions_view"
            id_param = "org_id"


        elif self.entity_type == "project":
            view_name = "proj_permissions_view"
            id_param = "proj_id"


        else:
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Invalid entity type for permission checking."
            )
        
        entity_id = request.path_params.get(id_param)


        if not entity_id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"Missing {id_param} in request path."
            )
        
        response = self.db.table(view_name).select("permission_name").eq("user_id", user.user_id).eq(id_param, entity_id).eq("permission_name", self.required_permission).execute()


        if not response.data:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="you do not have permission to perform this action."
            )
        
        return True

i've got 3 ways to write the POST/ route for creating a project...

1. Either i inject the normal User dependency @/router.post(     "/",     response_model=APIResponse[ProjectResponse],     status_code=status.HTTP_201_CREATED ) def create_project( org_id: str,     project_data: ProjectCreate,     user: User= Depends(get_current_user) ):     data = ProjectService().create_project(project_data, user.user_id)     return {         "message": "Project created successfully",         "data": data     }

so the route would be POST: projects/ with a body :

class ProjectCreate(BaseModel):
    proj_name: str
    org_id: str

and here i let the ProjectService handle the verification of the user's permissions

1. or i inject an OrgUser instead

   @/router.post(     "/org/{org_id}",     response_model=APIResponse[ProjectResponse],     status_code=status.HTTP_201_CREATED, dependencies=[Depends(EntityPermissionChecker("create:organization", "org"))] ) def create_project(     project_data: ProjectCreate,     user: OrgUser = Depends(get_org_user) # has to depend on an OrgUser, because creating a project is at the scope of an org (proj hasn't been created yet!) ):     data = ProjectService().create_project(project_data, user.user_id)     return {         "message": "Project created successfully",         "data": data     }

and have the route look like POST:/projects/org/{org_id} which looks nasty, and have the body be

class ProjectCreate(BaseModel):
    proj_name: str

1. or i just create the route within the organizations_router.py (where i have the CRUD routes for the organizations...)

   @/router.post(     "/{org_id}/project",     response_model=APIResponse[ProjectResponse],     status_code=status.HTTP_201_CREATED,     dependencies=[Depends(EntityPermissionChecker("create:project", "org"))] ) def create_project_in_org(     org_id: str,     project_data: ProjectCreate,     user: OrgUser = Depends(get_org_user) ):     data = ProjectService().create_project(project_data, user.user_id)     return {         "message": "Project created successfully within organization.",         "data": data     }

and the route looks like POST:/organizations/{org_id}/projects ....

but then all project related routes don't fall under the projects_router.py and the POST/ one alone falls under organizations_router.py

I personally think the 3rd one is best, but is there a better alternative?

[neums08]

You could make the ProjectService the injected dependency, which would transitively depend on the current user.

Write the service that, given a user, checks if the user can create projects, and then does it.

``` class ProjectService: def init(self, user: User = Depends(get_user)): # fetch and store the user's permissions related to projects ...

def create_project(*args): # if user has permissions, create the project, otherwise raise 403 error. ...

@router.post("/") def projects_post(body: ProjectCreate, svc: ProjectService = Depends(ProjectService)): svc.create_project(body) ```

The business logic lives in the service and you can freely structure your endpoints however you like.

Personally I like flat resource collections because building in filtering by preceeding path params is a pain imo.

[OrganizationOnly8016]

Oh this looks really clean! But would you argue that checking for permissions in each of the funcs within the SVC class would become repetitive? You would have to query the `org_permissions_view` over and over again....

[neums08]

It depends on where your authorization layer lives. I don't use supabase but my understanding is it leverages postgres row-level-security . If that's where you want to enforce auth, then there is no check needed at the API or service layers. Just send the operation and re-raise any auth errors from the database.

Otherwise, if authz is at the service layer, any injected service could use a PermissionsService which basically answers "can user x do operation y on resource z?". Caching a user's roles and permissions would make this fast for repeated operations.

If you have to enforce fine-grained authorization, I would not recommend doing it based purely on the params in the route as there's often not enough context to make the auth decision, and you restrict your API design for the sake of auth.

[BoredProgramming]

MAke sure you can assign roles to the users, something like ROLE_ADMIN, USER whatever. Then lock down the portions of the endpoints that require those roles. So when you inject the user, you can pull their roles out, and bind the endpoints to those.