refactor(view): extract validation logic from CreateTemporaryUserView to

reduce cyclomatic complexity 🔨

api/helpers.py

@@ -0,0 +1,83 @@
+from rest_framework.response import Response
+from typing import Optional
+from datetime import date
+from django.utils import timezone
+import re
+
+
+ID_CARD_PATTERN = r"^[A-Z]{1,2}[0-9]{6,7}$"
+MOBILE_PATTERN = r"^[7|9][0-9]{6}$"
+ACCOUNT_NUMBER_PATTERN = r"^(7\d{12}|9\d{16})$"
+
+
+class ErrorMessages:
+    USERNAME_EXISTS = "Username already exists."
+    MOBILE_EXISTS = "Mobile number already exists."
+    INVALID_ID_CARD = "Please enter a valid ID card number."
+    ID_CARD_EXISTS = "ID card already exists."
+    INVALID_MOBILE = "Please enter a valid mobile number."
+    INVALID_ACCOUNT = "Please enter a valid account number."
+    UNDERAGE_ERROR = "You must be 18 and above to signup."
+
+
+def validate_required_fields(data) -> Optional[Response]:
+    required_fields = {
+        "firstname": "First name",
+        "lastname": "Last name",
+        "username": "Username",
+        "address": "Address",
+        "mobile": "Mobile number",
+        "acc_no": "Account number",
+        "id_card": "ID card",
+        "dob": "Date of birth",
+        "atoll": "Atoll",
+        "island": "Island",
+    }
+
+    for field, label in required_fields.items():
+        if not data.get(field):
+            return Response({"message": f"{label} is required."}, status=400)
+
+    if data.get("terms_accepted") is None:
+        return Response({"message": "Terms acceptance is required."}, status=400)
+    if data.get("policy_accepted") is None:
+        return Response({"message": "Policy acceptance is required."}, status=400)
+
+    return None
+
+
+from .models import TemporaryUser, User
+
+def validate_unique_fields(username, mobile, id_card) -> Optional[Response]:
+    if mobile and (TemporaryUser.objects.filter(t_mobile=mobile).exists() or User.objects.filter(mobile=mobile).exists()):
+        return Response({"message": ErrorMessages.MOBILE_EXISTS}, status=400)
+
+    if username and (TemporaryUser.objects.filter(t_username=username).exists() or User.objects.filter(username=username).exists()):
+        return Response({"message": ErrorMessages.USERNAME_EXISTS}, status=400)
+
+    if id_card and (TemporaryUser.objects.filter(t_id_card=id_card).exists() or User.objects.filter(id_card=id_card).exists()):
+        return Response({"message": ErrorMessages.ID_CARD_EXISTS}, status=400)
+
+    return None
+
+
+
+
+def validate_patterns(id_card, mobile, acc_no) -> Optional[Response]:
+    if id_card and not re.match(ID_CARD_PATTERN, id_card):
+        return Response({"message": ErrorMessages.INVALID_ID_CARD}, status=400)
+
+    if mobile is None or not re.match(MOBILE_PATTERN, mobile):
+        return Response({"message": ErrorMessages.INVALID_MOBILE}, status=400)
+
+    if acc_no is None or not re.match(ACCOUNT_NUMBER_PATTERN, acc_no):
+        return Response({"message": ErrorMessages.INVALID_ACCOUNT}, status=400)
+
+    return None
+
+
+
+
+def calculate_age(dob: date) -> int:
+    today = timezone.now().date()
+    return today.year - dob.year - ((today.month, today.day) < (dob.month, dob.day))

api/views.py

@@ -2,6 +2,7 @@
 from django.contrib.auth import login
 
 # rest_framework imports
+from django.core.exceptions import ObjectDoesNotExist
 from rest_framework import generics, permissions
 from rest_framework.authtoken.serializers import AuthTokenSerializer
 from api.filters import UserFilter
@@ -22,19 +23,20 @@ from api.serializers import (
 )
 from django.shortcuts import get_object_or_404
 from django.utils import timezone
-from datetime import timedelta
 
 # knox imports
 from knox.views import LoginView as KnoxLoginView
 from knox.models import AuthToken
 from django_filters.rest_framework import DjangoFilterBackend
-import re
 from typing import cast, Dict, Any
 from django.core.mail import send_mail
 from django.db.models import Q
 from api.notifications import send_otp
 from .utils import check_person_api_verification
 import uuid
+from .helpers import ErrorMessages, validate_required_fields, validate_unique_fields, validate_patterns, calculate_age
+
+
 
 # local apps import
 from .serializers import (
@@ -45,19 +47,6 @@ from .serializers import (
     UserProfileUpdateSerializer,
 )
 
-ID_CARD_PATTERN = r"^[A-Z]{1,2}[0-9]{6,7}$"
-MOBILE_PATTERN = r"^[7|9][0-9]{6}$"
-ACCOUNT_NUMBER_PATTERN = r"^(7\d{12}|9\d{16})$"
-
-
-class ErrorMessages:
-    USERNAME_EXISTS = "Username already exists."
-    MOBILE_EXISTS = "Mobile number already exists."
-    INVALID_ID_CARD = "Please enter a valid ID card number."
-    ID_CARD_EXISTS = "ID card already exists."
-    INVALID_MOBILE = "Please enter a valid mobile number."
-    INVALID_ACCOUNT = "Please enter a valid account number."
-    UNDERAGE_ERROR = "You must be 18 and above to signup."
 
 
 @api_view(["GET"])
@@ -65,141 +54,94 @@ def healthcheck(request):
     return Response({"status": "Good"}, status=status.HTTP_200_OK)
 
 
+
 class CreateTemporaryUserView(generics.CreateAPIView):
-    # Create user API view
     serializer_class = TemporaryUserSerializer
     permission_classes = (permissions.AllowAny,)
     queryset = TemporaryUser.objects.all()
     throttle_classes = []
 
     def post(self, request, *args, **kwargs):
-        # Extract required fields from request data
+        # Extract data once
-        username = request.data.get("username")
+        data = request.data
-        address = request.data.get("address")
-        mobile = request.data.get("mobile")
-        acc_no = request.data.get("acc_no")
-        id_card = request.data.get("id_card")
-        dob = request.data.get("dob")
-        atoll_id = request.data.get("atoll")
-        island_id = request.data.get("island")
-        terms_accepted = request.data.get("terms_accepted")
-        policy_accepted = request.data.get("policy_accepted")
-        firstname = request.data.get("firstname")
-        lastname = request.data.get("lastname")
 
-        current_date = timezone.now()
+        # Validate required fields
+        required_error = validate_required_fields(data)
+        if required_error:
+            return required_error
+
+        # Parse DOB
+        dob_str = data.get("dob")
         try:
-            dob = timezone.datetime.strptime(str(dob), "%Y-%m-%d").date()
+            dob = timezone.datetime.strptime(str(dob_str), "%Y-%m-%d").date()  # pyright: ignore[reportAttributeAccessIssue]
         except ValueError:
-            return Response(
+            return Response({"message": "Invalid date format for DOB. Use YYYY-MM-DD."}, status=400)
-                {"message": "Invalid date format for DOB. Use YYYY-MM-DD."}, status=400
-            )
 
-        age_from_dob = (
+        # Check age
-            current_date.year
+        age = calculate_age(dob)
-            - dob.year
+        if age < 18:
-            - ((current_date.month, current_date.day) < (dob.month, dob.day))
-        )
-
-        if age_from_dob < 18:
             return Response({"message": ErrorMessages.UNDERAGE_ERROR}, status=400)
 
-        if (
+        # Validate uniqueness
-            TemporaryUser.objects.filter(t_mobile=mobile).exists()
+        uniqueness_error = validate_unique_fields(
-            or User.objects.filter(mobile=mobile).exists()
+            username=data.get("username"),
-        ):
+            mobile=data.get("mobile"),
-            return Response({"message": ErrorMessages.MOBILE_EXISTS}, status=400)
+            id_card=data.get("id_card"),
-        if (
+        )
-            TemporaryUser.objects.filter(t_username=username).exists()
+        if uniqueness_error:
-            or User.objects.filter(username=username).exists()
+            return uniqueness_error
-        ):
-            return Response({"message": ErrorMessages.USERNAME_EXISTS}, status=400)
-        if (
-            TemporaryUser.objects.filter(t_id_card=id_card).exists()
-            or User.objects.filter(id_card=id_card).exists()
-        ):
-            return Response({"message": "ID card already exists."}, status=400)
-        if (
-            TemporaryUser.objects.filter(t_id_card=id_card).exists()
-            or User.objects.filter(id_card=id_card).exists()
-        ):
-            return Response({"message": ErrorMessages.ID_CARD_EXISTS}, status=400)
-        if id_card and not re.match(ID_CARD_PATTERN, id_card):
-            return Response({"message": ErrorMessages.INVALID_ID_CARD}, status=400)
-        if mobile is None or not re.match(MOBILE_PATTERN, mobile):
-            return Response({"message": ErrorMessages.INVALID_MOBILE}, status=400)
-        if acc_no is None or not re.match(ACCOUNT_NUMBER_PATTERN, acc_no):
-            return Response({"message": ErrorMessages.INVALID_ACCOUNT}, status=400)
 
-        # Validate required fields first
+        # Validate patterns
-        validation_error = self.validate_required_fields(request.data)
+        pattern_error = validate_patterns(
-        if validation_error:
+            id_card=data.get("id_card"),
-            return validation_error
+            mobile=data.get("mobile"),
+            acc_no=data.get("acc_no"),
+        )
+        if pattern_error:
+            return pattern_error
 
-        # Fetch Atoll and Island instances
+        # Fetch related objects
+        atoll_id = data.get("atoll")
+        island_id = data.get("island")
         try:
             atoll = Atoll.objects.get(id=atoll_id)
             island = Island.objects.get(id=island_id)
-        except Atoll.DoesNotExist:
+        except ObjectDoesNotExist as e:
-            return Response({"message": "Atoll not found."}, status=404)
+            model_name = "Atoll" if isinstance(e, Atoll.DoesNotExist) else "Island"
-        except Island.DoesNotExist:
+            return Response({"message": f"{model_name} not found."}, status=404)
-            return Response({"message": "Island not found."}, status=404)
 
         # Create user
         temp_user = TemporaryUser.objects.create(
-            t_first_name=firstname,
+            t_first_name=data.get("firstname"),
-            t_last_name=lastname,
+            t_last_name=data.get("lastname"),
-            t_username=str(username),
+            t_username=str(data.get("username")),
             t_email=None,
-            t_address=address,
+            t_address=data.get("address"),
-            t_mobile=mobile,
+            t_mobile=data.get("mobile"),
-            t_acc_no=acc_no,
+            t_acc_no=data.get("acc_no"),
-            t_id_card=id_card,
+            t_id_card=data.get("id_card"),
             t_dob=dob,
             t_atoll=atoll,
             t_island=island,
-            t_terms_accepted=terms_accepted,
+            t_terms_accepted=data.get("terms_accepted"),
-            t_policy_accepted=policy_accepted,
+            t_policy_accepted=data.get("policy_accepted"),
         )
-        otp_expiry = timezone.now() + timedelta(minutes=3)
+
+        # Generate and send OTP
+        otp_expiry = timezone.now() + timezone.timedelta(minutes=3) #type: ignore
         formatted_time = otp_expiry.strftime("%d/%m/%Y %H:%M:%S")
         otp = temp_user.generate_otp()
         send_otp(
             str(temp_user.t_mobile),
             f"Your Registration SARLink OTP: {otp}. \nExpires at {formatted_time}. \n\n- SAR Link",
         )
+
+        # Return success
         serializer = self.get_serializer(temp_user)
         headers = self.get_success_headers(serializer.data)
         return Response(
             serializer.data, status=status.HTTP_201_CREATED, headers=headers
         )
-
-    def validate_required_fields(self, data):
-        required_fields = {
-            "firstname": "First name",
-            "lastname": "Last name",
-            "username": "Username",
-            "address": "Address",
-            "mobile": "Mobile number",
-            "acc_no": "Account number",
-            "id_card": "ID card",
-            "dob": "Date of birth",
-            "atoll": "Atoll",
-            "island": "Island",
-        }
-
-        for field, label in required_fields.items():
-            if not data.get(field):
-                return Response({"message": f"{label} is required."}, status=400)
-
-        if data.get("terms_accepted") is None:
-            return Response({"message": "Terms acceptance is required."}, status=400)
-        if data.get("policy_accepted") is None:
-            return Response({"message": "Policy acceptance is required."}, status=400)
-
-        return None
-
-
 class VerifyOTPView(generics.GenericAPIView):
     permission_classes = (permissions.AllowAny,)
     serializer_class = OTPVerificationSerializer
@@ -402,7 +344,7 @@ class KnoxTokenListApiView(
 
 class ListUserView(StaffEditorPermissionMixin, generics.ListAPIView):
     serializer_class = CustomReadOnlyUserSerializer
-    filter_backends = [DjangoFilterBackend]
+    filter_backends = [DjangoFilterBackend] #type: ignore
     filterset_fields = "__all__"
     filterset_class = UserFilter
     queryset = User.objects.all()
@@ -577,7 +519,7 @@ class UserDetailAPIView(StaffEditorPermissionMixin, generics.RetrieveAPIView):
         if (
             user != instance
             and not getattr(user, "is_admin", False)
-            and not user.is_superuser
+            and not user.is_superuser #type: ignore
         ):
             return Response(
                 {"message": "You are not authorized to view this user's details."},