feat: Implement team members endpoint for admin console

Author: rodmgwgu

openedx_authz/rest_api/utils.py

@@ -1,11 +1,26 @@
 """Utility functions for the Open edX AuthZ REST API."""
 
+from dataclasses import dataclass
+import logging
+
 from django.contrib.auth import get_user_model
 from django.db.models import Q
 
-from openedx_authz.api.data import GLOBAL_SCOPE_WILDCARD, ScopeData
+from openedx_authz.api.data import (
+    GLOBAL_SCOPE_WILDCARD,
+    ContentLibraryData,
+    CourseOverviewData,
+    OrgContentLibraryGlobData,
+    OrgCourseOverviewGlobData,
+    RoleAssignmentData,
+    ScopeData,
+)
+from openedx_authz.api.users import is_user_allowed
+from openedx_authz.constants.permissions import COURSES_VIEW_COURSE, VIEW_LIBRARY
 from openedx_authz.rest_api.data import SearchField, SortField, SortOrder
 
+logger = logging.getLogger(__name__)
+
 User = get_user_model()
 
 
@@ -135,3 +150,51 @@ def filter_users(users: list[dict], search: str | None, roles: list[str] | None)
         filtered_users.append(user)
 
     return filtered_users
+
+
+@dataclass
+class UserAssignments:
+    user: "User"
+    assignments: list[RoleAssignmentData]
+
+
+def get_user_assignment_map(role_assignments: list[RoleAssignmentData]) -> list[UserAssignments]:
+    """
+    Group role assignments by user
+    """
+    usernames = {assignment.subject.username for assignment in role_assignments}
+    user_map = get_user_map(usernames)
+
+    users_with_assignments: list[UserAssignments] = []
+
+    for username, user in user_map.items():
+        assignments = [a for a in role_assignments if a.subject.username == username]
+        users_with_assignments.append(UserAssignments(user=user, assignments=assignments))
+
+    return users_with_assignments
+
+
+def filter_allowed_assignments(
+    user: "User", assignments: list[RoleAssignmentData]
+) -> list[RoleAssignmentData]:
+    """
+    Filter the given role assignments to only include those that the user has permission to view.
+    """
+    allowed_assignments: list[RoleAssignmentData] = []
+    for assignment in assignments:
+        permission = None
+
+        # For CourseOverviewData and ContentLibraryData, check for the view permission
+        if isinstance(assignment.scope, (CourseOverviewData, OrgCourseOverviewGlobData)):
+            permission = COURSES_VIEW_COURSE.identifier
+        elif isinstance(assignment.scope, (ContentLibraryData, OrgContentLibraryGlobData)):
+            permission = VIEW_LIBRARY.identifier
+
+        if permission and is_user_allowed(
+            user_external_key=user.username,
+            action_external_key=permission,
+            scope_external_key=assignment.scope.external_key
+        ):
+            allowed_assignments.append(assignment)
+
+    return allowed_assignments

openedx_authz/rest_api/v1/fields.py

@@ -13,6 +13,17 @@ def to_internal_value(self, data):
     def to_representation(self, value):
         """Convert list to string separated by commas"""
         return ",".join(value).lower()
+    
+class CaseSensitiveCommaSeparatedListField(serializers.CharField):
+    """Serializer for a comma-separated list of strings, case-sensitive."""
+
+    def to_internal_value(self, data):
+        """Convert string separated by commas to list of unique items preserving order"""
+        return list(dict.fromkeys(item.strip() for item in data.split(",") if item.strip()))
+
+    def to_representation(self, value):
+        """Convert list to string separated by commas"""
+        return ",".join(value)
 
 
 class LowercaseCharField(serializers.CharField):

openedx_authz/rest_api/v1/serializers.py

@@ -5,8 +5,8 @@
 
 from openedx_authz import api
 from openedx_authz.rest_api.data import SortField, SortOrder
-from openedx_authz.rest_api.utils import get_generic_scope
-from openedx_authz.rest_api.v1.fields import CommaSeparatedListField, LowercaseCharField
+from openedx_authz.rest_api.utils import UserAssignments, get_generic_scope
+from openedx_authz.rest_api.v1.fields import CaseSensitiveCommaSeparatedListField, CommaSeparatedListField, LowercaseCharField
 
 User = get_user_model()
 
@@ -207,8 +207,8 @@ def get_roles(self, obj: api.RoleAssignmentData) -> list[str]:
 class ListTeamMembersSerializer(serializers.Serializer):  # pylint: disable=abstract-method
     """Serializer for listing team members."""
 
-    scopes = CommaSeparatedListField(required=False, default=[])
-    orgs = CommaSeparatedListField(required=False, default=[])
+    scopes = CaseSensitiveCommaSeparatedListField(required=False, default=[])
+    orgs = CaseSensitiveCommaSeparatedListField(required=False, default=[])
     sort_by = serializers.ChoiceField(
         required=False,
         choices=[(e.value, e.name) for e in SortField],
@@ -220,3 +220,27 @@ class ListTeamMembersSerializer(serializers.Serializer):  # pylint: disable=abst
         default=SortOrder.ASC,
     )
     search = LowercaseCharField(required=False, default=None)
+
+class TeamMemberSerializer(serializers.Serializer):  # pylint: disable=abstract-method
+    """Serializer for team members."""
+
+    username = serializers.SerializerMethodField()
+    full_name = serializers.SerializerMethodField()
+    email = serializers.SerializerMethodField()
+    assignation_count = serializers.SerializerMethodField()
+
+    def get_username(self, obj: UserAssignments) -> str:
+        """Get the username for the given role assignment."""
+        return getattr(obj.user, "username", "") if obj.user else ""
+
+    def get_full_name(self, obj: UserAssignments) -> str:
+        """Get the full name for the given role assignment."""
+        return obj.user.get_full_name() if obj.user else ""
+
+    def get_email(self, obj: UserAssignments) -> str:
+        """Get the email for the given role assignment."""
+        return getattr(obj.user, "email", "") if obj.user else ""
+
+    def get_assignation_count(self, obj: UserAssignments) -> int:
+        """Get the assignation count for the given role assignment."""
+        return len(obj.assignments)

openedx_authz/rest_api/v1/views.py

@@ -5,7 +5,6 @@
 permissions, roles, and user assignments within Open edX platform.
 """
 
-from dataclasses import dataclass
 import logging
 
 import edx_api_doc_tools as apidocs
@@ -17,11 +16,14 @@
 
 from openedx_authz import api
 from openedx_authz.constants import permissions
-from openedx_authz.rest_api.data import RoleOperationError, RoleOperationStatus
+from openedx_authz.rest_api.data import RoleOperationError, RoleOperationStatus, SortField
 from openedx_authz.rest_api.decorators import authz_permissions, view_auth_classes
 from openedx_authz.rest_api.utils import (
+    UserAssignments,
+    filter_allowed_assignments,
     filter_users,
     get_generic_scope,
+    get_user_assignment_map,
     get_user_by_username_or_email,
     get_user_map,
     sort_users,
@@ -37,6 +39,7 @@
     PermissionValidationResponseSerializer,
     PermissionValidationSerializer,
     RemoveUsersFromRoleWithScopeSerializer,
+    TeamMemberSerializer,
     UserRoleAssignmentSerializer,
 )
 
@@ -452,27 +455,6 @@ def get(self, request: HttpRequest) -> Response:
         serialized_data = ListRolesWithScopeResponseSerializer(paginated_response_data, many=True)
         return paginator.get_paginated_response(serialized_data.data)
 
-@dataclass
-class UserAssignments:
-    user: "User"
-    assignments: list[api.RoleAssignmentData]
-
-def get_user_assignment_map(role_assignments: list[api.RoleAssignmentData]) -> list[UserAssignments]:
-    """
-    Group role assignments by user
-    """
-    usernames = {assignment.subject.username for assignment in role_assignments}
-    user_map = get_user_map(usernames)
-
-    users_with_assignments: list[UserAssignments] = []
-
-    for username, user in user_map.items():
-        assignments = [a for a in role_assignments if a.subject.username == username]
-        users_with_assignments.append(UserAssignments(user=user, assignments=assignments))
-
-    return users_with_assignments
-
-
 
 @view_auth_classes()
 class TeamMembersAPIView(APIView):
@@ -507,31 +489,42 @@ def get(self, request: HttpRequest) -> Response:
         query_params = serializer.validated_data
 
         user_role_assignments = api.get_all_user_role_assignments()
+        user_role_assignments = filter_allowed_assignments(user=request.user, assignments=user_role_assignments)
         # Group assignments by user
         users_with_assignments = get_user_assignment_map(user_role_assignments)
 
-        if query_params.scopes:
-            # Filter by scopes
-            scopes = {s.strip() for s in query_params.scopes.split(",")}
-            users_with_assignments = [
-                uwa for uwa in users_with_assignments if any(a.scope.external_key in scopes for a in uwa.assignments)
-            ]
+        scopes = query_params.get('scopes')
+        orgs = query_params.get('orgs')
+        search = query_params.get('search')
+        order = query_params.get('order')
+        sort_by = query_params.get('sort_by', SortField.USERNAME)
 
-        if query_params.orgs:
+        if scopes:
+            # Filter by scopes
+            filtered_users: list[UserAssignments] = []
+            for uwa in users_with_assignments:
+                if any(a.scope.external_key in scopes for a in uwa.assignments):
+                    # Also filter assignments to reflect the correct number of assignments
+                    filtered_assignments = [a for a in uwa.assignments if a.scope.external_key in scopes]  
+                    filtered_users.append(UserAssignments(user=uwa.user, assignments=filtered_assignments))
+            users_with_assignments = filtered_users
+
+        if orgs:
             # Filter by orgs
-            orgs = {o.strip() for o in query_params.orgs.split(",")}
-            users_with_assignments = [
-                uwa for uwa in users_with_assignments if any(a.scope.org in orgs for a in uwa.assignments)
-            ]
-
-
-        logger.info(f">>>>Query params: {query_params}")
-        logger.info(f">>>>assignments: {user_role_assignments}")
-
-        # sort
-        # paginate
-
-
-        return Response({"results": []}, status=status.HTTP_200_OK)
-
-
+            filtered_users: list[UserAssignments] = []
+            for uwa in users_with_assignments:
+                if any(a.scope.org in orgs for a in uwa.assignments):
+                    # Also filter assignments to reflect the correct number of assignments
+                    filtered_assignments = [a for a in uwa.assignments if a.scope.org in orgs]
+                    filtered_users.append(UserAssignments(user=uwa.user, assignments=filtered_assignments))
+            users_with_assignments = filtered_users
+
+        team_members = TeamMemberSerializer(users_with_assignments, many=True)
+        # Search
+        filtered_team_members = filter_users(users=team_members.data, search=search, roles=None)
+        # Sort
+        sorted_team_members = sort_users(users=filtered_team_members, sort_by=sort_by, order=order)
+        # Paginate
+        paginator = self.pagination_class()
+        paginated_response_data = paginator.paginate_queryset(sorted_team_members, request)
+        return paginator.get_paginated_response(paginated_response_data)