Merge pull request #1648 from gtech-mulearn:dev

[FEAT] Discord like role management

api/dashboard/roles/dash_roles_serializer.py

@@ -4,7 +4,60 @@
 
 from db.user import Role, User, UserRoleLink
 from utils.permission import JWTUtils
-from utils.utils import DateTimeUtils
+from utils.utils import DateTimeUtils, DiscordWebhooks
+from utils.types import WebHookActions, WebHookCategory
+from django.db.models import Q
+from django.db import transaction
+
+class UserRoleLinkManagementSerializer(serializers.ModelSerializer):
+    """
+    Serializer used by UserRoleLinkManagement API to lists the
+    details of the user with a specific role
+    """
+
+    class Meta:
+        model = User
+        fields = ["id", "muid", "fullname"]
+
+
+class RoleAssignmentSerializer(serializers.Serializer):
+    """
+    Used by UserRoleLinkManagement to assign
+    a role to a large number of users
+    """
+
+    role = serializers.PrimaryKeyRelatedField(queryset=Role.objects.all())
+    users = serializers.ListField(child=serializers.UUIDField())
+    created_by = serializers.PrimaryKeyRelatedField(queryset=User.objects.all())
+
+    def validate(self, attrs):
+        data = super().validate(attrs)
+        attrs = set(attrs["users"])
+        users = User.objects.filter(
+            ~Q(user_role_link_user__role=data["role"]), pk__in=attrs
+        )
+        if users.count() != len(attrs):
+            raise serializers.ValidationError("One or more user IDs are invalid.")
+        else:
+            data["users"] = users
+            return data
+
+    def create(self, validated_data):
+        users = validated_data.pop("users")
+        validated_data["created_at"] = DateTimeUtils.get_current_utc_time()
+        validated_data["verified"] = True
+        user_roles_to_create = [
+            UserRoleLink(user=user, **validated_data) for user in users
+        ]
+        with transaction.atomic():
+            UserRoleLink.objects.bulk_create(user_roles_to_create)
+            DiscordWebhooks.general_updates(
+                WebHookCategory.BULK_ROLE.value,
+                WebHookActions.UPDATE.value,
+                validated_data["role"].title,
+                ",".join(list(users.values_list("id", flat=True))),
+            )
+        return user_roles_to_create
 
 
 class RoleDashboardSerializer(serializers.ModelSerializer):
@@ -49,18 +102,16 @@ def create(self, validated_data):
         return super().create(validated_data)
 
 
-
-
 class UserRoleSearchSerializer(serializers.ModelSerializer):
     class Meta:
         model = User
         fields = ["id", "fullname", "muid"]
 
 
 class UserRoleCreateSerializer(serializers.ModelSerializer):
-
     user_id = serializers.CharField(required=True, source="user.id")
     role_id = serializers.CharField(required=True, source="role.id")
+
     class Meta:
         model = UserRoleLink
         fields = ["user_id", "role_id"]
@@ -80,11 +131,13 @@ def create(self, validated_data):
         validated_data["created_at"] = DateTimeUtils.get_current_utc_time()
 
         return super().create(validated_data)
-
+
+
 class UserRoleBulkAssignSerializer(serializers.ModelSerializer):
     user_id = serializers.CharField(required=True)
     role_id = serializers.CharField(required=True)
     created_by_id = serializers.CharField(required=True, allow_null=False)
+
     class Meta:
         model = UserRoleLink
         fields = [
@@ -99,7 +152,6 @@ class Meta:
     def to_representation(self, instance):
         representation = super().to_representation(instance)
 
-        representation['user_id'] = instance.user.fullname if instance.user else None
-        representation['role_id'] = instance.role.title if instance.role else None
+        representation["user_id"] = instance.user.fullname if instance.user else None
+        representation["role_id"] = instance.role.title if instance.role else None
         return representation
-

api/dashboard/roles/dash_roles_views.py

@@ -13,6 +13,8 @@
 from tempfile import NamedTemporaryFile
 from io import BytesIO
 from django.http import FileResponse
+from django.db.models import Q
+
 
 class RoleAPI(APIView):
     authentication_classes = [CustomizePermission]
@@ -156,6 +158,66 @@ def get(self, request, role_id):
         ).get_success_response()
 
 
+class UserRoleLinkManagement(APIView):
+    authentication_classes = [CustomizePermission]
+    """
+    This API is creates an interface to help manage the user and role link
+    by providing support for 
+    - Listing all users with the given role
+    - Giving a lot of users a specific role
+    """
+
+    @role_required([RoleType.ADMIN.value])
+    def get(self, request, role_id):
+        """
+        Lists all the users with a given role
+        """
+        users = (
+            User.objects.filter(user_role_link_user__role__pk=role_id)
+            .distinct()
+            .all()
+        )
+        serialized_users = dash_roles_serializer.UserRoleLinkManagementSerializer(
+            users, many=True
+        )
+        return CustomResponse(response=serialized_users.data).get_success_response()
+
+    @role_required([RoleType.ADMIN.value])
+    def put(self, request, role_id):
+        """
+        Lists all the users without a given role;
+        used to assign roles
+        """
+        users = (
+            User.objects.filter(~Q(user_role_link_user__role__pk=role_id))
+            .distinct()
+            .all()
+        )
+        serialized_users = dash_roles_serializer.UserRoleLinkManagementSerializer(
+            users, many=True
+        )
+        return CustomResponse(response=serialized_users.data).get_success_response()
+
+    @role_required([RoleType.ADMIN.value])
+    def patch(self, request):
+        """
+        Assigns a large bunch of users a certain role
+        """
+        request_data = request.data.copy()
+        request_data[
+            "created_by"
+        ] = JWTUtils.fetch_user_id(request)
+        serialized_users = dash_roles_serializer.RoleAssignmentSerializer(
+            data=request_data
+        )
+        if serialized_users.is_valid():
+            serialized_users.save()
+            return CustomResponse(
+                general_message="Successfully gave all users the requested role"
+            ).get_success_response()
+        return CustomResponse(response=serialized_users.errors).get_failure_response()
+
+
 class UserRole(APIView):
     authentication_classes = [CustomizePermission]
 
@@ -208,36 +270,39 @@ def delete(self, request):
             general_message="User Role deleted successfully"
         ).get_success_response()
 
+
 class RoleBaseTemplateAPI(APIView):
     authentication_classes = [CustomizePermission]
 
     def get(self, request):
-        wb = load_workbook('./api/dashboard/roles/assets/role_base_template.xlsx')
-        ws = wb['Data Definitions']
+        wb = load_workbook("./api/dashboard/roles/assets/role_base_template.xlsx")
+        ws = wb["Data Definitions"]
 
-        roles = Role.objects.all().values_list('title', flat=True)
-        data = {
-            'role': roles
-        }
+        roles = Role.objects.all().values_list("title", flat=True)
+        data = {"role": roles}
         # Write data column-wise
         for col_num, (col_name, col_values) in enumerate(data.items(), start=1):
             for row, value in enumerate(col_values, start=2):
                 ws.cell(row=row, column=col_num, value=value)
-        
+
         # Save the file
         with NamedTemporaryFile() as tmp:
-            tmp.close() # with statement opened tmp, close it so wb.save can open it
+            tmp.close()  # with statement opened tmp, close it so wb.save can open it
             wb.save(tmp.name)
-            with open(tmp.name, 'rb') as f:
+            with open(tmp.name, "rb") as f:
                 f.seek(0)
                 new_file_object = f.read()
-        return FileResponse(BytesIO(new_file_object), as_attachment=True, filename='role_base_template.xlsx')
-
+        return FileResponse(
+            BytesIO(new_file_object),
+            as_attachment=True,
+            filename="role_base_template.xlsx",
+        )
+
+
 class UserRoleBulkAssignAPI(APIView):
     authentication_classes = [CustomizePermission]
 
     @role_required([RoleType.ADMIN.value])
-
     def post(self, request):
         try:
             file_obj = request.FILES["user_roles_list"]
@@ -254,17 +319,14 @@ def post(self, request):
                 general_message="Empty csv file."
             ).get_failure_response()
 
-        temp_headers = [
-            "muid",
-            "role"
-        ]
+        temp_headers = ["muid", "role"]
         first_entry = excel_data[0]
         for key in temp_headers:
             if key not in first_entry:
                 return CustomResponse(
                     general_message=f"{key} does not exist in the file."
                 ).get_failure_response()
-            
+
         excel_data = [row for row in excel_data if any(row.values())]
         valid_rows = []
         error_rows = []
@@ -276,7 +338,7 @@ def post(self, request):
         for row in excel_data[1:]:
             keys_to_keep = ["muid", "role"]
             row_keys = list(row.keys())
-            
+
             # Remove columns other than "muid" and "role"
             for key in row_keys:
                 if key not in keys_to_keep:
@@ -294,30 +356,27 @@ def post(self, request):
             else:
                 user_role_link_to_check.add((user, role))
 
-        users = User.objects.filter(
-            muid__in=users_to_fetch
-            ).values(
-                "id",
-                "muid",
-            )
-        roles = Role.objects.filter(
-            title__in=roles_to_fetch
-            ).values(
-                "id",
-                "title",
-            )
-        existing_user_role_links = list(UserRoleLink.objects.filter(
-            user__muid__in=users_to_fetch, 
-            role__title__in=roles_to_fetch
-            ).values_list('user__muid', 'role__title'))
+        users = User.objects.filter(muid__in=users_to_fetch).values(
+            "id",
+            "muid",
+        )
+        roles = Role.objects.filter(title__in=roles_to_fetch).values(
+            "id",
+            "title",
+        )
+        existing_user_role_links = list(
+            UserRoleLink.objects.filter(
+                user__muid__in=users_to_fetch, role__title__in=roles_to_fetch
+            ).values_list("user__muid", "role__title")
+        )
         users_dict = {user["muid"]: user["id"] for user in users}
         roles_dict = {role["title"]: role["id"] for role in roles}
 
         for row in excel_data[1:]:
             user = row.pop("muid")
             role = row.pop("role")
 
-            user_id = users_dict.get(user) 
+            user_id = users_dict.get(user)
             role_id = roles_dict.get(role)
 
             if not user_id:
@@ -344,19 +403,22 @@ def post(self, request):
                 row["created_by_id"] = request_user_id
                 valid_rows.append(row)
 
-        user_roles_serializer = dash_roles_serializer.UserRoleBulkAssignSerializer(data=valid_rows, many=True)
+        user_roles_serializer = dash_roles_serializer.UserRoleBulkAssignSerializer(
+            data=valid_rows, many=True
+        )
         success_data = []
         if user_roles_serializer.is_valid():
             user_roles_serializer.save()
             for user_role_data in user_roles_serializer.data:
-                    success_data.append({
-                        'user': user_role_data.get('user_id', ''),
-                        'role': user_role_data.get('role_id', ''),
-                })
+                success_data.append(
+                    {
+                        "user": user_role_data.get("user_id", ""),
+                        "role": user_role_data.get("role_id", ""),
+                    }
+                )
         else:
             error_rows.append(user_roles_serializer.errors)
 
         return CustomResponse(
             response={"Success": success_data, "Failed": error_rows}
         ).get_success_response()
-

api/dashboard/roles/urls.py

@@ -6,7 +6,9 @@
 urlpatterns = [
     path('user-role/<str:role_id>/', dash_roles_views.UserRoleSearchAPI.as_view(), name='search-user-role'),
     path('base-template/', dash_roles_views.RoleBaseTemplateAPI.as_view(), name="roles-base-template"),
-    path('bulk-assign/', dash_roles_views.UserRoleBulkAssignAPI.as_view(), name="user-roles-import"),
+    path('bulk-assign/', dash_roles_views.UserRoleLinkManagement.as_view(), name="user-roles-assign"), # used to assign a bunch of users a role
+    path('bulk-assign/<str:role_id>/', dash_roles_views.UserRoleLinkManagement.as_view(), name="user-roles-list"), # used to get the list of users to assign a role
+    path('bulk-assign-excel/', dash_roles_views.UserRoleBulkAssignAPI.as_view(), name="user-roles-assign-excel"),
     path('user-role/', dash_roles_views.UserRole.as_view(), name='create-delete-user-role'),
     path('', dash_roles_views.RoleAPI.as_view(), name="roles-list"),
     path('', dash_roles_views.RoleAPI.as_view(), name="roles-create"),

db/user.py

@@ -81,6 +81,9 @@ class UserRoleLink(models.Model):
     class Meta:
         managed = False
         db_table = 'user_role_link'
+        constraints = [
+            models.UniqueConstraint(fields=['role', 'user'], name="UserToRole")
+        ]
 
 
 class Socials(models.Model):

utils/types.py

@@ -78,6 +78,7 @@ class WebHookCategory(Enum):
     USER = 'user'
     USER_NAME = 'user-name'
     USER_PROFILE = 'user-profile'
+    BULK_ROLE = 'bulk-role'
 
 
 class RefferalType(Enum):