squash!: Refactored to extern most of the filtering logic to an api function

Author: rodmgwgu

openedx_authz/api/data.py

@@ -1189,3 +1189,10 @@ def __repr__(self):
 class UserAssignments:
     user: "User"
     assignments: list[RoleAssignmentData]
+
+
+class UserAssignmentsFilter(Enum):
+    """Enum for the filters that can be applied over UserAssignments."""
+
+    SCOPES = "scopes"
+    ORGS = "orgs"

openedx_authz/api/users.py

@@ -11,10 +11,16 @@
 
 from openedx_authz.api.data import (
     ActionData,
+    ContentLibraryData,
+    CourseOverviewData,
+    OrgContentLibraryGlobData,
+    OrgCourseOverviewGlobData,
     PermissionData,
     RoleAssignmentData,
     RoleData,
     ScopeData,
+    UserAssignments,
+    UserAssignmentsFilter,
     UserData,
 )
 from openedx_authz.api.permissions import is_subject_allowed
@@ -33,18 +39,21 @@
     unassign_role_from_subject_in_scope,
     unassign_subject_from_all_roles,
 )
+from openedx_authz.api.utils import filter_user_assignments, get_user_assignment_map
+from openedx_authz.constants.permissions import COURSES_VIEW_COURSE, VIEW_LIBRARY
 
 __all__ = [
     "assign_role_to_user_in_scope",
     "batch_assign_role_to_users_in_scope",
     "unassign_role_from_user",
     "batch_unassign_role_from_users",
-    "get_all_user_role_assignments",
     "get_user_role_assignments",
     "get_user_role_assignments_in_scope",
     "get_user_role_assignments_for_role_in_scope",
     "get_user_role_assignments_filtered",
+    "get_all_user_role_assignments",
     "get_all_user_role_assignments_in_scope",
+    "get_all_user_role_assignments_by_user_filtered",
     "is_user_allowed",
     "get_scopes_for_user_and_permission",
     "get_users_for_role_in_scope",
@@ -120,15 +129,6 @@ def batch_unassign_role_from_users(users: list[str], role_external_key: str, sco
     )
 
 
-def get_all_user_role_assignments() -> list[RoleAssignmentData]:
-    """Get all roles for all users across all scopes.
-
-    Returns:
-        list[RoleAssignmentData]: A list of role assignments and all their metadata assigned to the user.
-    """
-    return get_all_subject_role_assignments()
-
-
 def get_user_role_assignments(user_external_key: str) -> list[RoleAssignmentData]:
     """Get all roles for a user across all scopes.
 
@@ -202,6 +202,15 @@ def get_user_role_assignments_filtered(
     )
 
 
+def get_all_user_role_assignments() -> list[RoleAssignmentData]:
+    """Get all roles for all users across all scopes.
+
+    Returns:
+        list[RoleAssignmentData]: A list of role assignments and all their metadata assigned to the user.
+    """
+    return get_all_subject_role_assignments()
+
+
 def get_all_user_role_assignments_in_scope(
     scope_external_key: str,
 ) -> list[RoleAssignmentData]:
@@ -216,6 +225,71 @@ def get_all_user_role_assignments_in_scope(
     return get_all_subject_role_assignments_in_scope(ScopeData(external_key=scope_external_key))
 
 
+def _filter_allowed_assignments(
+    user_external_key: str, 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_external_key,
+            action_external_key=permission,
+            scope_external_key=assignment.scope.external_key,
+        ):
+            allowed_assignments.append(assignment)
+
+    return allowed_assignments
+
+
+def get_all_user_role_assignments_by_user_filtered(
+    orgs: list[str] = None,
+    scopes: list[str] = None,
+    allowed_for_user_external_key: str = None,
+) -> list[UserAssignments]:
+    """
+    Get all user role assignments filtered by orgs and/or scopes, and only include
+    assignments that the specified user has permission to view.
+
+    Args:
+        orgs: Optional list of orgs to filter by (e.g., ['edX', 'MITx']).
+        scopes: Optional list of scopes to filter by (e.g., ['lib:DemoX:CSPROB']).
+        allowed_for_user_external_key: The username to check permissions against (e.g., 'john_doe').
+
+    Returns:
+        list[UserAssignments]: A list of users with their role assignments, filtered by orgs/scopes and permissions.
+    """
+    user_role_assignments = get_all_user_role_assignments()
+    # Filter assignments based on the user's permissions
+    user_role_assignments = _filter_allowed_assignments(
+        user_external_key=allowed_for_user_external_key,
+        assignments=user_role_assignments,
+    )
+    # Group assignments by user
+    users_with_assignments = get_user_assignment_map(user_role_assignments)
+
+    users_with_assignments = filter_user_assignments(
+        users_with_assignments=users_with_assignments,
+        by=UserAssignmentsFilter.SCOPES,
+        values=scopes,
+    )
+    users_with_assignments = filter_user_assignments(
+        users_with_assignments=users_with_assignments,
+        by=UserAssignmentsFilter.ORGS,
+        values=orgs,
+    )
+    return users_with_assignments
+
+
 def is_user_allowed(
     user_external_key: str,
     action_external_key: str,

openedx_authz/api/utils.py

@@ -0,0 +1,77 @@
+"""Utility functions used on api"""
+
+from django.contrib.auth import get_user_model
+
+from openedx_authz.api.data import (
+    RoleAssignmentData,
+    UserAssignments,
+    UserAssignmentsFilter,
+)
+
+User = get_user_model()
+
+
+def get_user_map(usernames: list[str]) -> dict[str, User]:
+    """
+    Retrieve a dictionary mapping usernames to User objects for efficient batch lookups.
+
+    This function performs a single optimized database query to fetch multiple users,
+    making it ideal for scenarios where we need to look up several users at once
+    (e.g., when serializing multiple user role assignments).
+
+    Args:
+        usernames (list[str]): List of usernames to retrieve. Duplicates are automatically
+            handled by the database query.
+
+    Returns:
+        dict[str, User]: Dictionary mapping each username to its corresponding User object.
+            Only users that exist in the database are included in the returned dictionary.
+    """
+    users = User.objects.filter(username__in=usernames).select_related("profile")
+    return {user.username: user for user in users}
+
+
+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_user_assignments(
+    users_with_assignments: list[UserAssignments],
+    by: UserAssignmentsFilter,
+    values: list[str],
+) -> list[UserAssignments]:
+    """
+    Filter user assignments by orgs or scopes.
+    """
+    if not values:
+        return users_with_assignments
+
+    def _get_value_to_filter(assignment: RoleAssignmentData) -> str:
+        if by == UserAssignmentsFilter.SCOPES:
+            return assignment.scope.external_key
+        elif by == UserAssignmentsFilter.ORGS:
+            return assignment.scope.org
+        else:
+            raise ValueError(f"Invalid filter: '{by}'. Must be one of {[f.value for f in UserAssignmentsFilter]}")
+
+    filtered_users: list[UserAssignments] = []
+    for uwa in users_with_assignments:
+        if any(_get_value_to_filter(a) in values for a in uwa.assignments):
+            # Also filter assignments to reflect the correct number of assignments
+            filtered_assignments = [a for a in uwa.assignments if _get_value_to_filter(a) in values]
+            filtered_users.append(UserAssignments(user=uwa.user, assignments=filtered_assignments))
+    users_with_assignments = filtered_users
+
+    return filtered_users

openedx_authz/rest_api/data.py

@@ -50,10 +50,3 @@ class RoleOperationError(BaseEnum):
     USER_DOES_NOT_HAVE_ROLE = "user_does_not_have_role"
     ROLE_ASSIGNMENT_ERROR = "role_assignment_error"
     ROLE_REMOVAL_ERROR = "role_removal_error"
-
-
-class UserAssignmentsFilter(BaseEnum):
-    """Enum for the filters that can be applied over UserAssignments."""
-
-    SCOPES = "scopes"
-    ORGS = "orgs"

openedx_authz/rest_api/utils.py

@@ -2,26 +2,14 @@
 
 import logging
 
-from django.contrib.auth import get_user_model
-
 from openedx_authz.api.data import (
     GLOBAL_SCOPE_WILDCARD,
-    ContentLibraryData,
-    CourseOverviewData,
-    OrgContentLibraryGlobData,
-    OrgCourseOverviewGlobData,
-    RoleAssignmentData,
     ScopeData,
-    UserAssignments,
 )
-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, UserAssignmentsFilter
+from openedx_authz.rest_api.data import SearchField, SortField, SortOrder
 
 logger = logging.getLogger(__name__)
 
-User = get_user_model()
-
 
 def get_generic_scope(scope: ScopeData) -> ScopeData:
     """
@@ -45,26 +33,6 @@ def get_generic_scope(scope: ScopeData) -> ScopeData:
     return ScopeData(namespaced_key=f"{scope.NAMESPACE}{ScopeData.SEPARATOR}{GLOBAL_SCOPE_WILDCARD}")
 
 
-def get_user_map(usernames: list[str]) -> dict[str, User]:
-    """
-    Retrieve a dictionary mapping usernames to User objects for efficient batch lookups.
-
-    This function performs a single optimized database query to fetch multiple users,
-    making it ideal for scenarios where we need to look up several users at once
-    (e.g., when serializing multiple user role assignments).
-
-    Args:
-        usernames (list[str]): List of usernames to retrieve. Duplicates are automatically
-            handled by the database query.
-
-    Returns:
-        dict[str, User]: Dictionary mapping each username to its corresponding User object.
-            Only users that exist in the database are included in the returned dictionary.
-    """
-    users = User.objects.filter(username__in=usernames).select_related("profile")
-    return {user.username: user for user in users}
-
-
 def sort_users(
     users: list[dict],
     sort_by: SortField = SortField.USERNAME,
@@ -129,73 +97,3 @@ def filter_users(users: list[dict], search: str | None, roles: list[str] | None)
         filtered_users.append(user)
 
     return filtered_users
-
-
-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
-
-
-def filter_user_assignments(
-    users_with_assignments: list[UserAssignments],
-    by: UserAssignmentsFilter,
-    values: list[str],
-) -> list[UserAssignments]:
-    """
-    Filter user assignments by orgs or scopes.
-    """
-    if not values:
-        return users_with_assignments
-
-    def _get_value_to_filter(assignment: RoleAssignmentData) -> str:
-        if by == UserAssignmentsFilter.SCOPES:
-            return assignment.scope.external_key
-        elif by == UserAssignmentsFilter.ORGS:
-            return assignment.scope.org
-        else:
-            raise ValueError(f"Invalid filter: '{by}'. Must be one of {UserAssignmentsFilter.values()}")
-
-    filtered_users: list[UserAssignments] = []
-    for uwa in users_with_assignments:
-        if any(_get_value_to_filter(a) in values for a in uwa.assignments):
-            # Also filter assignments to reflect the correct number of assignments
-            filtered_assignments = [a for a in uwa.assignments if _get_value_to_filter(a) in values]
-            filtered_users.append(UserAssignments(user=uwa.user, assignments=filtered_assignments))
-    users_with_assignments = filtered_users
-
-    return filtered_users

openedx_authz/rest_api/v1/serializers.py

@@ -4,8 +4,9 @@
 from rest_framework import serializers
 
 from openedx_authz import api
+from openedx_authz.api.data import UserAssignments
 from openedx_authz.rest_api.data import SortField, SortOrder
-from openedx_authz.rest_api.utils import UserAssignments, get_generic_scope
+from openedx_authz.rest_api.utils import get_generic_scope
 from openedx_authz.rest_api.v1.fields import (
     CaseSensitiveCommaSeparatedListField,
     CommaSeparatedListField,