[MAJOR] Created a system to handle all unhandled exceptions

api/dashboard/profile/profile_serializer.py

@@ -8,6 +8,7 @@
 from db.organization import UserOrganizationLink
 from db.task import InterestGroup, KarmaActivityLog, Level, TaskList, Wallet, UserIgLink, UserLvlLink
 from db.user import User, UserSettings, Socials
+from utils.exception import CustomException
 from utils.permission import JWTUtils
 from utils.types import OrganizationType, RoleType, MainRoles
 from utils.utils import DateTimeUtils
@@ -323,7 +324,7 @@ def update(self, instance, validated_data):
                 for ig_data in ig_details
             ]
             if len(user_ig_links) > 3:
-                raise ValueError("Cannot add more than 3 interest groups")
+                raise CustomException("Cannot add more than 3 interest groups")
             UserIgLink.objects.bulk_create(user_ig_links)
             return super().update(instance, validated_data)
 
@@ -383,9 +384,9 @@ def create_karma_activity_log(task_title, karma_value):
             old_account_url = old_accounts[account]
             if account_url != old_account_url:
                 if old_account_url is None:
-                    create_karma_activity_log("social_" + account, 50)
+                    create_karma_activity_log(f"social_{account}", 50)
                 elif account_url is None:
-                    create_karma_activity_log("social_" + account, -50)
+                    create_karma_activity_log(f"social_{account}", -50)
 
         instance.save()
         return instance

api/dashboard/profile/profile_view.py

@@ -5,6 +5,7 @@
 from db.task import InterestGroup, KarmaActivityLog, Level
 from db.user import Role
 from db.user import User, UserSettings, UserRoleLink, Socials
+from utils.exception import CustomException
 from utils.permission import CustomizePermission, JWTUtils
 from utils.response import CustomResponse
 from utils.types import WebHookActions, WebHookCategory
@@ -86,7 +87,7 @@ def patch(self, request):
                 general_message="Interest Group edited successfully"
             ).get_success_response()
 
-        except ValueError as e:
+        except CustomException as e:
             return CustomResponse(general_message=str(e)).get_failure_response()
 
 

api/integrations/integrations_helper.py

@@ -7,6 +7,7 @@
 
 from db.integrations import Integration
 from mulearnbackend.settings import SECRET_KEY
+from utils.exception import CustomException
 from utils.response import CustomResponse
 
 
@@ -21,16 +22,19 @@ def get_authorization_id(token: str) -> str | None:
     :return: the authorization ID if the token is valid and has not expired. If the token is invalid or
     has expired, it returns None.
     """
-    payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
-    authorization_id = payload.get("authorization_id")
-    exp_timestamp = payload.get("exp", 0)
-
-    if exp_timestamp and datetime.now(pytz.utc) < datetime.fromtimestamp(
-        exp_timestamp, tz=pytz.utc
-    ):
-        return authorization_id
-    else:
-        raise ValueError("Token invalid or expired")
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
+        authorization_id = payload.get("authorization_id")
+        exp_timestamp = payload.get("exp", 0)
+
+        if exp_timestamp and datetime.now(pytz.utc) < datetime.fromtimestamp(
+            exp_timestamp, tz=pytz.utc
+        ):
+            return authorization_id
+        else:
+            raise CustomException("Token invalid or expired")
+    except jwt.ExpiredSignatureError:
+        raise CustomException("This token has expired, maybe again with a new one!")
 
 
 def generate_confirmation_token(authorization_id: str) -> str:
@@ -68,14 +72,14 @@ def wrapper(self, request, *args, **kwargs):
             try:
                 auth_header = request.META.get("HTTP_AUTHORIZATION")
                 if not auth_header or not auth_header.startswith("Bearer "):
-                    raise ValueError("Invalid Authorization header")
+                    raise CustomException("Invalid Authorization header")
 
                 token = auth_header.split(" ")[1]
 
                 if not Integration.objects.filter(
                     token=token, name=integration_name
                 ).first():
-                    raise ValueError("Invalid Authorization header")
+                    raise CustomException("Invalid Authorization header")
                 else:
                     result = func(self, request, *args, **kwargs)
                 return result
@@ -111,7 +115,7 @@ def get_access_token(
         ).json()
 
         if response.get("statusCode") != 200:
-            raise ValueError(
+            raise CustomException(
                 "Oops! The username or password didn't match our records. Please double-check and try again."
             )
     else:
@@ -120,7 +124,7 @@ def get_access_token(
         ).json()
 
         if response.get("statusCode") != 200:
-            raise ValueError(
+            raise CustomException(
                 "Oops! We couldn't find that account. Please double-check your details and try again."
             )
 
@@ -137,10 +141,10 @@ def get_access_token(
 def handle_response(response: dict) -> None:
     if response.get("statusCode") != 200:
         if "emailOrMuid" in response:
-            raise ValueError(
+            raise CustomException(
                 "Oops! The username or password didn't match our records. Please double-check and try again."
             )
         else:
-            raise ValueError(
+            raise CustomException(
                 "Oops! We couldn't find that account. Please double-check your details and try again."
             )

api/integrations/kkem/kkem_helper.py

@@ -9,6 +9,7 @@
 from Crypto.Util.Padding import unpad
 
 from db.integrations import Integration
+from utils.exception import CustomException
 from utils.types import IntegrationType
 from utils.utils import send_template_mail
 
@@ -31,7 +32,7 @@ def send_data_to_kkem(kkem_link):
     response_data = response.json()
 
     if not response_data["request_status"]:
-        raise ValueError("Invalid jsid")
+        raise CustomException("Invalid jsid")
 
     send_connection_successful_email(kkem_link.user)
     return response_data
@@ -65,11 +66,13 @@ def ensure_padding(encoded_str):
         try:
             decrypted = unpad(decrypted_data, AES.block_size)
         except:
-            raise ValueError("Invalid padding or incorrect key")
+            raise CustomException("Invalid padding or incorrect key")
 
         return parse_qs(decrypted.decode("utf-8"))
     except Exception as e:
-        raise ValueError("The given token seems to be invalid do re-check and try again!")
+        raise CustomException(
+            "The given token seems to be invalid do re-check and try again!"
+        )
 
 
 def send_connection_successful_email(user):

api/integrations/kkem/kkem_serializer.py

@@ -3,6 +3,7 @@
 
 from db.integrations import Integration, IntegrationAuthorization
 from db.user import User
+from utils.exception import CustomException
 from utils.types import IntegrationType
 from utils.utils import DateTimeUtils
 
@@ -28,8 +29,7 @@ class Meta:
     def get_total_karma(self, obj):
         karma = (
             obj.wallet_user.karma
-            if hasattr(obj, "wallet_user")
-            and hasattr(obj.wallet_user, "karma")
+            if hasattr(obj, "wallet_user") and hasattr(obj.wallet_user, "karma")
             else 0
         )
         return karma
@@ -117,21 +117,21 @@ def create(self, validated_data):
 
         if kkem_link:
             if (
-                    self.context["type"] == "login"
-                    and kkem_link.integration_value == jsid
-                    and kkem_link.verified
+                self.context["type"] == "login"
+                and kkem_link.integration_value == jsid
+                and kkem_link.verified
             ):
                 return kkem_link
 
             elif kkem_link.verified:
-                raise ValueError(
+                raise CustomException(
                     "Your μLearn account is already connected to a KKEM account"
                 )
 
             elif kkem_link.user == user:
                 self.update_integration(validated_data, kkem_link)
             else:
-                raise ValueError("Something went wrong")
+                raise CustomException("Something went wrong")
         else:
             kkem_link = self.create_kkem_link(
                 user, integration, dwms_id, jsid, validated_data["verified"]
@@ -142,24 +142,22 @@ def create(self, validated_data):
         return kkem_link
 
     def verify_user(self, user_muid):
-        if user := User.objects.filter(
-                Q(muid=user_muid) | Q(email=user_muid)
-        ).first():
+        if user := User.objects.filter(Q(muid=user_muid) | Q(email=user_muid)).first():
             return user
         else:
-            raise ValueError(
+            raise CustomException(
                 "Oops! We couldn't find that account. Please double-check your details and try again."
             )
 
     def get_kkem_link(self, user, integration, jsid):
         if kkem_link := IntegrationAuthorization.objects.filter(
-                user=user, integration=integration
+            user=user, integration=integration
         ).first():
             return kkem_link
         if IntegrationAuthorization.objects.filter(
-                integration_value=jsid, integration=integration
+            integration_value=jsid, integration=integration
         ).exists():
-            raise ValueError("This KKEM account is already connected to another user")
+            raise CustomException("This KKEM account is already connected to another user")
 
         return None
 

api/integrations/kkem/kkem_views.py

@@ -1,13 +1,14 @@
+import logging
 from datetime import datetime
 
-import logging
 import requests
 from django.db.models import F, Prefetch
 from rest_framework.views import APIView
 
 from db.integrations import Integration, IntegrationAuthorization
 from db.task import InterestGroup, KarmaActivityLog, TaskList, UserIgLink
 from db.user import User
+from utils.exception import CustomException
 from utils.response import CustomResponse
 from utils.types import IntegrationType
 from utils.utils import DateTimeUtils, send_template_mail
@@ -60,7 +61,7 @@ def get(self, request):
                 base_queryset = base_queryset.filter(
                     karma_activity_log_user__updated_at__gte=from_datetime
                 )
-            except ValueError:
+            except CustomException:
                 return CustomResponse(
                     general_message="Invalid datetime format",
                 ).get_failure_response()
@@ -122,10 +123,8 @@ def post(self, request):
                 general_message="We've set up your authorization! Please check your email for further instructions."
             ).get_success_response()
 
-        except ValueError as e:
-            return CustomResponse(
-                general_message=str(e)
-            ).get_failure_response()
+        except CustomException as e:
+            return CustomResponse(general_message=str(e)).get_failure_response()
 
     def patch(self, request, token):
         try:
@@ -150,10 +149,9 @@ def patch(self, request, token):
                 general_message="Invalid or missing Token"
             ).get_failure_response()
 
-        except ValueError as e:
-            return CustomResponse(
-                general_message=str(e)
-            ).get_failure_response()
+        except CustomException as e:
+            return CustomResponse(general_message=str(e)).get_failure_response()
+
 
 class KKEMIntegrationLogin(APIView):
     def post(self, request):
@@ -184,16 +182,11 @@ def post(self, request):
                 response["data"] = serialized_set.data
 
             return CustomResponse(
-                general_message=general_message,
-                response=response
+                general_message=general_message, response=response
             ).get_success_response()
 
-        except ValueError as e:
-            return CustomResponse(
-                general_message=str(e)
-            ).get_failure_response()
-
-
+        except CustomException as e:
+            return CustomResponse(general_message=str(e)).get_failure_response()
 
 
 class KKEMdetailsFetchAPI(APIView):
@@ -226,12 +219,8 @@ def get(self, request, encrypted_data):
 
             return CustomResponse(response=result_data).get_success_response()
 
-        except ValueError as e:
-            return CustomResponse(
-                general_message=str(e)
-            ).get_failure_response()
-
-
+        except CustomException as e:
+            return CustomResponse(general_message=str(e)).get_failure_response()
 
 
 class KKEMUserStatusAPI(APIView):
@@ -242,9 +231,5 @@ def get(self, request, encrypted_data):
                 response={"muid": details["mu_id"][0] if "mu_id" in details else None}
             ).get_success_response()
 
-        except ValueError as e:
-            return CustomResponse(
-                general_message=str(e)
-            ).get_failure_response()
-
-
+        except CustomException as e:
+            return CustomResponse(general_message=str(e)).get_failure_response()