feat: add PolicyBulkViewSet and PolicySingleViewSet for managing bulk and single policy operations

Author: BryanttV

openedx_authz/urls.py

@@ -2,14 +2,16 @@
 URLs for openedx_authz.
 """
 
-from django.urls import re_path, include
+from django.urls import include, re_path
 from rest_framework.routers import DefaultRouter
-from .views import LibraryViewSet, AdminRoleAssignmentViewSet, UserPermissionViewSet
+
+from .views import AdminRoleAssignmentViewSet, LibraryViewSet, PolicyBulkViewSet, PolicySingleViewSet
 
 router = DefaultRouter()
 router.register(r"libraries", LibraryViewSet, basename="library")
 router.register(r"admin-roles", AdminRoleAssignmentViewSet, basename="admin-roles")
-router.register(r"user-permissions", UserPermissionViewSet, basename="user-permissions")
+router.register(r"policy-bulk", PolicyBulkViewSet, basename="policy-bulk")
+router.register(r"policy-single", PolicySingleViewSet, basename="policy-single")
 
 urlpatterns = [
     re_path(r"^api/", include(router.urls)),

openedx_authz/views.py

@@ -123,7 +123,6 @@ def create(self, request):
         }
         ```
         """
-        enforcer.enable_auto_save(True)
         username = request.data["username"]
         enforcer.add_role_for_user(username, "admin")
         return Response(f"Admin role assigned to user {username}", status=status.HTTP_201_CREATED)
@@ -138,86 +137,125 @@ def destroy(self, request, pk=None):
         return Response(f"Admin role removed from user {username}", status=status.HTTP_204_NO_CONTENT)
 
 
-class UserPermissionViewSet(viewsets.ViewSet):
+class PolicyBulkViewSet(viewsets.ViewSet):
     """
-    ViewSet for managing specific user permissions using Casbin.
-    Allows adding or removing specific permissions for users on resources.
-
-    Example:
-    ```json
-    {
-        "username": "john_doe",
-        "obj": "/api/libraries/",
-        "act": "GET"
-    }
-    ```
+    ViewSet for bulk policy operations using Casbin's add_policies and remove_policies.
+    This is a simple testing interface for bulk policy management.
     """
 
     def create(self, request):
         """
-        POST /user-permissions/
-        Add a specific permission to a user.
+        POST /policy-bulk/
+        Add multiple policies at once using add_policies.
 
         Example request body:
         ```json
         {
-            "username": "john_doe",
-            "obj": "/api/libraries/123/",
-            "act": "GET"
+            "policies": [
+                ["user1", "/api/resource1/", "GET"],
+                ["user2", "/api/resource2/", "POST"],
+                ["user3", "/api/resource3/", "DELETE"]
+            ]
         }
         ```
         """
-        username = request.data.get("username")
-        obj = request.data.get("obj")
-        act = request.data.get("act")
-
-        if not all([username, obj, act]):
-            return Response({"error": "username, obj, and act are required fields"}, status=status.HTTP_400_BAD_REQUEST)
+        policies = request.data.get("policies", [])
+        result = enforcer.add_policies(policies)
+        return Response(
+            {
+                "message": "Bulk policy addition completed",
+                "success": result,
+                "policies_added": len(policies),
+                "policies": policies,
+            },
+            status=status.HTTP_201_CREATED if result else status.HTTP_200_OK,
+        )
 
-        enforcer.add_policy(username, obj, act)
+    def destroy(self, request, pk=None):  # pylint: disable=unused-argument
+        """
+        DELETE /policy-bulk/remove/
+        Remove multiple policies at once using remove_policies.
+        Uses request body instead of pk since we need to pass multiple policies.
 
+        Example request body:
+        ```json
+        {
+            "policies": [
+                ["user1", "/api/resource1/", "GET"],
+                ["user2", "/api/resource2/", "POST"]
+            ]
+        }
+        ```
+        """
+        policies = request.data.get("policies", [])
+        result = enforcer.remove_policies(policies)
         return Response(
             {
-                "message": f"Permission '{act}' on '{obj}' granted to user '{username}'",
-                "username": username,
-                "obj": obj,
-                "act": act,
+                "message": "Bulk policy removal completed",
+                "success": result,
+                "policies_removed": len(policies),
+                "policies": policies,
             },
-            status=status.HTTP_201_CREATED,
+            status=status.HTTP_204_NO_CONTENT if result else status.HTTP_200_OK,
         )
 
-    def destroy(self, request, pk=None):
+
+class PolicySingleViewSet(viewsets.ViewSet):
+    """
+    ViewSet for single policy operations using Casbin's add_policy and remove_policy.
+    Simple testing interface for individual policy management.
+    """
+
+    def create(self, request):
+        """
+        POST /policy-single/
+        Add a single policy using add_policy.
+
+        Example request body:
+        ```json
+        {
+            "subject": "user1",
+            "object": "/api/resource1/",
+            "action": "GET"
+        }
+        ```
         """
-        DELETE /user-permissions/{username}/
-        Remove a specific permission from a user.
+        subject = request.data.get("subject")
+        obj = request.data.get("object")
+        action = request.data.get("action")
+        result = enforcer.add_policy(subject, obj, action)
+        return Response(
+            {
+                "message": "Policy added successfully" if result else "Policy already exists",
+                "success": result,
+                "policy": [subject, obj, action],
+            },
+            status=status.HTTP_201_CREATED if result else status.HTTP_200_OK,
+        )
 
-        Query parameters:
-        - obj: The resource path (required)
-        - act: The action/method (required)
+    def destroy(self, request, pk=None):  # pylint: disable=unused-argument
+        """
+        DELETE /policy-single/remove/
+        Remove a single policy using remove_policy.
 
-        Example: DELETE /user-permissions/john_doe/?obj=/api/libraries/123/&act=GET
+        Example request body:
+        ```json
+        {
+            "subject": "user1",
+            "object": "/api/resource1/",
+            "action": "GET"
+        }
+        ```
         """
-        username = pk
-        obj = request.query_params.get("obj")
-        act = request.query_params.get("act")
-
-        if not all([obj, act]):
-            return Response({"error": "obj and act query parameters are required"}, status=status.HTTP_400_BAD_REQUEST)
-
-        result = enforcer.remove_policy(username, obj, act)
-
-        if result:
-            return Response(
-                {
-                    "message": f"Permission '{act}' on '{obj}' removed from user '{username}'",
-                    "username": username,
-                    "obj": obj,
-                    "act": act,
-                },
-                status=status.HTTP_204_NO_CONTENT,
-            )
-        else:
-            return Response(
-                {"error": f"Permission not found for user '{username}' on '{obj}' with action '{act}'"},
-                status=status.HTTP_404_NOT_FOUND,
-            )
+        subject = request.data.get("subject")
+        obj = request.data.get("object")
+        action = request.data.get("action")
+        result = enforcer.remove_policy(subject, obj, action)
+        return Response(
+            {
+                "message": "Policy removed successfully" if result else "Policy not found",
+                "success": result,
+                "policy": [subject, obj, action],
+            },
+            status=status.HTTP_204_NO_CONTENT if result else status.HTTP_404_NOT_FOUND,
+        )