feat: add twillio functionality and task queue

Author: obaraelijah

app/user/manager.py app/user/managers.pyapp/user/manager.py renamed to app/user/managers.py

@@ -2,3 +2,94 @@
 from datetime import datetime, timezone
 
 
+from core.models import AuditableModel
+from django.conf import settings
+from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+from django.contrib.postgres.fields import ArrayField
+from .enums import TOKEN_TYPE_CHOICE, ROLE_CHOICE
+from .managers import CustomUserManager
+
+
+def default_role():
+ return ["CUSTOMER"]
+
+
+class User(AbstractBaseUser, PermissionsMixin):
+ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+ email = models.EmailField(
+ _("email address"), null=True, blank=True, unique=True)
+ password = models.CharField(max_length=255, null=True)
+ firstname = models.CharField(max_length=255, blank=True, null=True)
+ lastname = models.CharField(max_length=255, blank=True, null=True)
+ image = models.FileField(upload_to="users/", blank=True, null=True)
+ phone = models.CharField(max_length=30, unique=True, blank=True, null=True)
+ is_locked = models.BooleanField(default=False)
+ is_staff = models.BooleanField(default=False)
+ is_active = models.BooleanField(default=False)
+ is_admin = models.BooleanField(default=False)
+ last_login = models.DateTimeField(null=True, blank=True)
+ created_at = models.DateTimeField(auto_now_add=True)
+ updated_at = models.DateTimeField(auto_now=True)
+ verified = models.BooleanField(default=False)
+ USERNAME_FIELD = "phone"
+ REQUIRED_FIELDS = []
+ objects = CustomUserManager()
+ roles = ArrayField(models.CharField(max_length=20, blank=True,
+ choices=ROLE_CHOICE), default=default_role, size=6)
+
+ class Meta:
+ ordering = ("-created_at",)
+
+ def __str__(self) -> str:
+ return self.phone
+
+ def save_last_login(self) -> None:
+ self.last_login = datetime.now()
+ self.save()
+
+
+class PendingUser(AuditableModel):
+ phone = models.CharField(max_length=20)
+ verification_code = models.CharField(max_length=8, blank=True, null=True)
+ password = models.CharField(max_length=255, null=True)
+
+
+ def __str__(self):
+ return f"{str(self.phone)} {self.verification_code}"
+ 
+ def is_valid(self) -> bool:
+ """10 mins OTP validation"""
+ lifespan_in_seconds = float(settings.OTP_EXPIRE_TIME * 60)
+ now = datetime.now(timezone.utc)
+ time_diff = now - self.created_at
+ time_diff = time_diff.total_seconds()
+ if time_diff >= lifespan_in_seconds:
+ return False
+ return True
+
+
+class Token(models.Model):
+ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+ user = models.ForeignKey(settings.AUTH_USER_MODEL,
+ on_delete=models.CASCADE)
+ token = models.CharField(max_length=8)
+ token_type = models.CharField(max_length=100, choices=TOKEN_TYPE_CHOICE)
+ created_at = models.DateTimeField(auto_now_add=True)
+
+ def __str__(self):
+ return f"{str(self.user)} {self.token}"
+
+ def is_valid(self) -> bool:
+ lifespan_in_seconds = float(settings.TOKEN_LIFESPAN * 60 )
+ now = datetime.now(timezone.utc)
+ time_diff = now - self.created_at
+ time_diff = time_diff.total_seconds()
+ if time_diff >= lifespan_in_seconds:
+ return False
+ return True
+
+ def reset_user_password(self, password: str) -> None:
+ self.user.set_password(password)
+ self.user.save()

app/user/models.py

@@ -0,0 +1,9 @@
+from core.celery import APP
+from .utils import send_sms
+#celery for asynchronous processing, to send phone notifications asynchronously.
+
+
+@APP.task()
+def send_phone_notification(user_data):
+ send_sms(user_data['message'], user_data['phone'])
+ 

app/user/task.py

@@ -0,0 +1,61 @@
+import base64
+import os
+import re
+import pyotp
+from django.conf import settings
+from rest_framework import permissions, serializers
+from twilio.rest import Client
+
+from .enums import SystemRoleEnum
+from .models import User
+
+client = Client(settings.TWILIO_ACCOUNT_SID, settings.TWILIO_AUTH_TOKEN)
+
+def get_user_role_names(user:User)->list:
+ """
+ Returns a list of role names for the given user.
+ """
+ return user.roles.values_list('name', flat=True)
+
+def is_admin_user(user:User)->bool:
+ """
+ Check an authenticated user is an admin or not
+ """
+ return user.is_admin or SystemRoleEnum.ADMIN in user.roles
+
+
+class IsAdmin(permissions.BasePermission):
+ """Allows access only to Admin users."""
+ message = "Only Admins are authorized to perform this action."
+ 
+ def has_permission(self, request, view):
+ if not request.user.is_authenticated:
+ return False
+ return is_admin_user(request.user)
+ 
+
+def send_sms(message:str, phone:str):
+ client.messages.create(
+ body=message,
+ from_=settings.TWILIO_PHONE_NUMBER,
+ to=phone
+ )
+ return
+
+
+def clean_phone(number:str):
+ """Validates number start with +254 or 0, then 10 digits"""
+ number_pattern = re.compile(r'^(?:\+234|0)\d{10}$')
+ result = number_pattern.match(number)
+ if result:
+ if number.startswith('0'): 
+ return '+254' + number[1:]
+ return number
+ else:
+ raise serializers.ValidationError({'phone': 'Incorrect phone number.'}) 
+ 
+
+def generate_otp()->int:
+ totp = pyotp.TOTP(base64.b32encode(os.urandom(16)).decode('utf-8'))
+ otp = totp.now()
+ return otp