Merge pull request #1386 from gtech-mulearn:dev

[FEAT] Configure CustomException Error handling

api/dashboard/roles/dash_roles_views.py

@@ -6,6 +6,7 @@
 from utils.response import CustomResponse
 from utils.types import RoleType, WebHookActions, WebHookCategory
 from utils.utils import CommonUtils, DiscordWebhooks
+
 from . import dash_roles_serializer
 
 
@@ -38,32 +39,23 @@ def get(self, request):
             },
         )
         serializer = dash_roles_serializer.RoleDashboardSerializer(
-            queryset.get("queryset"),
-            many=True
+            queryset.get("queryset"), many=True
         )
 
         return CustomResponse().paginated_response(
-            data=serializer.data,
-            pagination=queryset.get("pagination")
+            data=serializer.data, pagination=queryset.get("pagination")
         )
 
     @role_required([RoleType.ADMIN.value])
     def patch(self, request, roles_id):
-
-        role = Role.objects.filter(id=roles_id).first()
+        role = Role.objects.get(id=roles_id)
         old_name = role.title
 
         serializer = dash_roles_serializer.RoleDashboardSerializer(
-            role,
-            data=request.data,
-            partial=True,
-            context={
-                "request": request
-            }
+            role, data=request.data, partial=True, context={"request": request}
         )
 
         if not serializer.is_valid():
-
             return CustomResponse(
                 general_message=serializer.errors
             ).get_failure_response()
@@ -73,10 +65,7 @@ def patch(self, request, roles_id):
             newname = role.title
 
             DiscordWebhooks.general_updates(
-                WebHookCategory.ROLE.value,
-                WebHookActions.EDIT.value,
-                newname,
-                old_name
+                WebHookCategory.ROLE.value, WebHookActions.EDIT.value, newname, old_name
             )
 
             return CustomResponse(
@@ -94,23 +83,16 @@ def delete(self, request, roles_id):
         role.delete()
 
         DiscordWebhooks.general_updates(
-            WebHookCategory.ROLE.value,
-            WebHookActions.DELETE.value,
-            role.title
+            WebHookCategory.ROLE.value, WebHookActions.DELETE.value, role.title
         )
         return CustomResponse(
             general_message="Role deleted successfully"
         ).get_success_response()
 
     @role_required([RoleType.ADMIN.value])
     def post(self, request):
-
         serializer = dash_roles_serializer.RoleDashboardSerializer(
-            data=request.data,
-            partial=True,
-            context={
-                "request": request
-            }
+            data=request.data, partial=True, context={"request": request}
         )
 
         if serializer.is_valid():
@@ -126,9 +108,7 @@ def post(self, request):
                 response={"data": serializer.data}
             ).get_success_response()
 
-        return CustomResponse(
-            general_message=serializer.errors
-        ).get_failure_response()
+        return CustomResponse(general_message=serializer.errors).get_failure_response()
 
 
 class RoleManagementCSV(APIView):
@@ -139,40 +119,27 @@ def get(self, request):
         role = Role.objects.all()
 
         role_serializer_data = dash_roles_serializer.RoleDashboardSerializer(
-            role,
-            many=True
+            role, many=True
         ).data
-        return CommonUtils.generate_csv(
-            role_serializer_data,
-            "Roles"
-        )
+        return CommonUtils.generate_csv(role_serializer_data, "Roles")
 
 
 class UserRoleSearchAPI(APIView):
     authentication_classes = [CustomizePermission]
 
     @role_required([RoleType.ADMIN.value])
     def get(self, request, role_id):
-        user = User.objects.filter(
-            user_role_link_user__role_id=role_id
-        ).distinct()
+        user = User.objects.filter(user_role_link_user__role_id=role_id).distinct()
 
         paginated_queryset = CommonUtils.get_paginated_queryset(
             user,
             request,
-            [
-                "muid", "first_name", "last_name"
-            ],
-            {
-                "muid": "muid",
-                "first_name": "first_name",
-                "last_name": "last_name"
-            },
+            ["muid", "first_name", "last_name"],
+            {"muid": "muid", "first_name": "first_name", "last_name": "last_name"},
         )
 
         serializer = dash_roles_serializer.UserRoleSearchSerializer(
-            paginated_queryset.get("queryset"),
-            many=True
+            paginated_queryset.get("queryset"), many=True
         ).data
 
         return CustomResponse(
@@ -188,12 +155,8 @@ class UserRole(APIView):
 
     @role_required([RoleType.ADMIN.value])
     def post(self, request):
-
         serializer = dash_roles_serializer.UserRoleCreateSerializer(
-            data=request.data,
-            context={
-                "request": request
-            }
+            data=request.data, context={"request": request}
         )
 
         if not serializer.is_valid():
@@ -214,12 +177,8 @@ def post(self, request):
 
     @role_required([RoleType.ADMIN.value])
     def delete(self, request):
-
         serializer = dash_roles_serializer.UserRoleCreateSerializer(
-            data=request.data,
-            context={
-                "request": request
-            }
+            data=request.data, context={"request": request}
         )
 
         if not serializer.is_valid():
@@ -230,10 +189,7 @@ def delete(self, request):
         user_id = request.data.get("user_id")
         role_id = request.data.get("role_id")
 
-        user_role_link = UserRoleLink.objects.get(
-            role_id=role_id,
-            user_id=user_id
-        )
+        user_role_link = UserRoleLink.objects.get(role_id=role_id, user_id=user_id)
 
         user_role_link.delete()
 
@@ -245,4 +201,3 @@ def delete(self, request):
         return CustomResponse(
             general_message="User Role deleted successfully"
         ).get_success_response()
-

mulearnbackend/middlewares.py

@@ -1,15 +1,19 @@
 import hmac
+import json
 import logging
+import traceback
 
 import decouple
+from django.conf import settings
 from django.http import JsonResponse
 from rest_framework import status
 from rest_framework.renderers import JSONRenderer
 
+from utils.exception import CustomException
 from utils.response import CustomResponse
 from utils.utils import _CustomHTTPHandler
 
-logger = logging.getLogger('django')
+logger = logging.getLogger("django")
 
 
 class IpBindingMiddleware(object):
@@ -69,17 +73,76 @@ def __call__(self, request):
 
 
 class UniversalErrorHandlerMiddleware:
+    """
+    Middleware for handling exceptions and generating error responses.
+
+    Args:
+        get_response: The callable that takes a request and returns a response.
+
+    Methods:
+        __call__(self, request): Process the request and return the response.
+        log_exception(self, request, exception): Log the exception and request information.
+        process_exception(self, request, exception): Process the exception and return a response.
+
+    """
+
     def __init__(self, get_response):
         self.get_response = get_response
 
     def __call__(self, request):
         return self.get_response(request)
 
+    def log_exception(self, request, exception):
+        """
+        Log the exception and prints the information in CLI if DEBUG is True.
+
+        Args:
+            request: The request object.
+            exception: The exception object.
+
+        """
+        error_message = (
+            f"Exception Type: {type(exception).__name__}; "
+            f"Exception Message: {str(exception)}; "
+            f"Traceback: {traceback.format_exc()}"
+        )
+        logger.error(error_message)
+
+        if settings.DEBUG:
+            print(error_message)
+
+        request_info = (
+            f"Request Info: METHOD: {request.method}; \n"
+            f"\tPATH: {request.path}; \n"
+            f"\tDATA: {json.loads(request.body.decode('utf-8')) if hasattr(request, 'body') else 'No Data'}\n"
+        )
+        logger.error(request_info)
+
+        # Print to terminal if DEBUG is True
+        if settings.DEBUG:
+            print(request_info)
+
     def process_exception(self, request, exception):
-        logger.error(str(exception))
-        response = CustomResponse(
-            general_message="Something went wrong"
-        ).get_failure_response()
+        """
+        Process the exception and return a response.
+
+        Args:
+            request: The request object.
+            exception: The exception object.
+
+        Returns:
+            A response object.
+
+        """
+        if isinstance(exception, CustomException):
+            response = CustomResponse(
+                general_message=exception.detail,
+            ).get_failure_response(status_code=exception.status_code)
+        else:
+            self.log_exception(request, exception)
+            response = CustomResponse(
+                general_message="Something went wrong"
+            ).get_failure_response()
 
         # Set the renderer and renderer context
         renderer = JSONRenderer()

utils/exception.py

@@ -1,7 +1,13 @@
 from rest_framework.serializers import ValidationError
 
 
-class CustomException(ValidationError):
+class CustomException(Exception):
+    def __init__(self, detail="Something went wrong", status_code=403):
+        self.detail = detail
+        self.status_code = status_code
+
+
+class UnauthorizedAccessException(ValidationError):
     def __init__(self, detail="Something went wrong", status_code=403):
         self.detail = detail
         self.status_code = status_code

utils/permission.py

@@ -11,7 +11,7 @@
 
 from mulearnbackend.settings import SECRET_KEY
 from utils.utils import DateTimeUtils
-from .exception import CustomException
+from .exception import UnauthorizedAccessException
 from .response import CustomResponse
 
 from db.user import DynamicRole, DynamicUser
@@ -49,7 +49,7 @@ def authenticate(self, request):
             tuple: A tuple of (user, token_payload) if authentication is successful.
 
         Raises:
-            CustomException: If authentication fails.
+            UnauthorizedAccessException: If authentication fails.
         """
         return JWTUtils.is_jwt_authenticated(request)
 
@@ -113,41 +113,41 @@ def is_jwt_authenticated(request):
         try:
             auth_header = get_authorization_header(request).decode("utf-8")
             if not auth_header or not auth_header.startswith(token_prefix):
-                raise CustomException("Invalid token header")
+                raise UnauthorizedAccessException("Invalid token header")
 
             token = auth_header[len(token_prefix):].strip()
             if not token:
-                raise CustomException("Empty Token")
+                raise UnauthorizedAccessException("Empty Token")
 
             payload = jwt.decode(token, secret_key, algorithms=["HS256"], verify=True)
 
             user_id = payload.get("id")
             expiry = datetime.strptime(payload.get("expiry"), "%Y-%m-%d %H:%M:%S%z")
 
             if not user_id or expiry < DateTimeUtils.get_current_utc_time():
-                raise CustomException("Token Expired or Invalid")
+                raise UnauthorizedAccessException("Token Expired or Invalid")
 
             return None, payload
         except jwt.exceptions.InvalidSignatureError as e:
-            raise CustomException(
+            raise UnauthorizedAccessException(
                 {
                     "hasError": True,
                     "message": {"general": [str(e)]},
                     "statusCode": 1000,
                 }
             ) from e
         except jwt.exceptions.DecodeError as e:
-            raise CustomException(
+            raise UnauthorizedAccessException(
                 {
                     "hasError": True,
                     "message": {"general": [str(e)]},
                     "statusCode": 1000,
                 }
             ) from e
         except AuthenticationFailed as e:
-            raise CustomException(str(e)) from e
+            raise UnauthorizedAccessException(str(e)) from e
         except Exception as e:
-            raise CustomException(
+            raise UnauthorizedAccessException(
                 {
                     "hasError": True,
                     "message": {"general": [str(e)]},