mirror of
https://github.com/HamzaSha1/zod-backend.git
synced 2025-07-15 10:05:21 +00:00
sonar fixes
This commit is contained in:
@ -1,6 +1,9 @@
|
||||
"""Account app file"""
|
||||
"""Import Django"""
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class AccountConfig(AppConfig):
|
||||
"""default configurations"""
|
||||
default_auto_field = 'django.db.models.BigAutoField'
|
||||
name = 'account'
|
||||
|
@ -1,6 +1,8 @@
|
||||
"""Account module file"""
|
||||
"""Django import"""
|
||||
from django.db import models
|
||||
import random
|
||||
from django.contrib.auth.models import User
|
||||
"""App import"""
|
||||
from base.constants import USER_TYPE
|
||||
# Create your models here.
|
||||
|
||||
|
@ -1,25 +1,21 @@
|
||||
"""Account serializer"""
|
||||
"""Django Imoprt"""
|
||||
from rest_framework import serializers
|
||||
from django.contrib.auth.models import User
|
||||
from rest_framework_simplejwt.tokens import RefreshToken
|
||||
"""App import"""
|
||||
from guardian.models import Guardian
|
||||
from junior.models import Junior
|
||||
from account.models import UserProfile, UserEmailOtp, UserPhoneOtp, DefaultTaskImages
|
||||
from base.constants import GUARDIAN, JUNIOR, SUPERUSER
|
||||
from django.db import transaction
|
||||
from base.messages import ERROR_CODE_REQUIRED, ERROR_CODE, SUCCESS_CODE, STATUS_CODE_ERROR
|
||||
from django.core.exceptions import ObjectDoesNotExist
|
||||
from django.contrib.auth import authenticate
|
||||
from rest_framework import viewsets, status
|
||||
from rest_framework.decorators import action
|
||||
from django.contrib.auth import authenticate, login
|
||||
from rest_framework_simplejwt.tokens import RefreshToken
|
||||
from guardian.utils import upload_image_to_alibaba
|
||||
from .utils import get_token
|
||||
|
||||
|
||||
class GoogleLoginSerializer(serializers.Serializer):
|
||||
"""google login serializer"""
|
||||
access_token = serializers.CharField(max_length=5000, required=True)
|
||||
|
||||
class Meta:
|
||||
class Meta(object):
|
||||
"""meta class"""
|
||||
fields = ('access_token',)
|
||||
|
||||
@ -86,8 +82,7 @@ class ChangePasswordSerializer(serializers.Serializer):
|
||||
|
||||
def validate_current_password(self, value):
|
||||
user = self.context
|
||||
if self.context.password not in ('', None):
|
||||
if user.check_password(value):
|
||||
if self.context.password not in ('', None) and user.check_password(value):
|
||||
return value
|
||||
raise serializers.ValidationError(ERROR_CODE['2015'])
|
||||
def create(self, validated_data):
|
||||
@ -109,6 +104,7 @@ class ForgotPasswordSerializer(serializers.Serializer):
|
||||
email = serializers.EmailField()
|
||||
|
||||
class SuperUserSerializer(serializers.ModelSerializer):
|
||||
"""Super admin serializer"""
|
||||
user_type = serializers.SerializerMethodField('get_user_type')
|
||||
|
||||
def get_user_type(self, obj):
|
||||
|
@ -1,3 +1,5 @@
|
||||
"""Test cases file of account"""
|
||||
"""Django import"""
|
||||
from django.test import TestCase
|
||||
|
||||
# Create your tests here.
|
||||
|
@ -1,9 +1,9 @@
|
||||
""" Urls files"""
|
||||
"""Django import"""
|
||||
from django.urls import path, include
|
||||
from rest_framework.decorators import api_view
|
||||
"""Third party import"""
|
||||
from rest_framework import routers
|
||||
"""Import view functions"""
|
||||
from .views import (UserLogin, SendPhoneOtp, UserPhoneVerification, UserEmailVerification, ReSendEmailOtp,
|
||||
ForgotPasswordAPIView, ResetPasswordAPIView, ChangePasswordAPIView, UpdateProfileImage,
|
||||
GoogleLoginViewSet, SigninWithApple, ProfileAPIViewSet, UploadImageAPIViewSet,
|
||||
@ -13,15 +13,23 @@ router = routers.SimpleRouter()
|
||||
|
||||
"""API End points with router"""
|
||||
router.register('user', UserLogin, basename='user')
|
||||
"""super admin login"""
|
||||
router.register('admin', UserLogin, basename='admin')
|
||||
"""google login end point"""
|
||||
router.register('google-login', GoogleLoginViewSet, basename='admin')
|
||||
# router.register('send-phone-otp', SendPhoneOtp, basename='send-phone-otp')
|
||||
# router.register('user-phone-verification', UserPhoneVerification, basename='user-phone-verification')
|
||||
router.register('send-phone-otp', SendPhoneOtp, basename='send-phone-otp')
|
||||
router.register('user-phone-verification', UserPhoneVerification, basename='user-phone-verification')
|
||||
"""email verification end point"""
|
||||
router.register('user-email-verification', UserEmailVerification, basename='user-email-verification')
|
||||
"""Resend email otp end point"""
|
||||
router.register('resend-email-otp', ReSendEmailOtp, basename='resend-email-otp')
|
||||
"""Profile end point"""
|
||||
router.register('profile', ProfileAPIViewSet, basename='profile')
|
||||
"""Upload default task image end point"""
|
||||
router.register('upload-default-task-image', UploadImageAPIViewSet, basename='upload-default-task-image')
|
||||
"""Fetch default task image end point"""
|
||||
router.register('default-task-image', DefaultImageAPIViewSet, basename='default-task-image')
|
||||
"""Define url pattern"""
|
||||
urlpatterns = [
|
||||
path('api/v1/', include(router.urls)),
|
||||
path('api/v1/forgot-password/', ForgotPasswordAPIView.as_view()),
|
||||
|
@ -1,8 +1,9 @@
|
||||
"""Account utils"""
|
||||
"""Third party Django app"""
|
||||
"""Import django"""
|
||||
from django.conf import settings
|
||||
from rest_framework import viewsets, status
|
||||
from rest_framework.response import Response
|
||||
"""Third party Django app"""
|
||||
from templated_email import send_templated_mail
|
||||
import jwt
|
||||
from datetime import datetime
|
||||
@ -11,6 +12,7 @@ from uuid import uuid4
|
||||
import secrets
|
||||
|
||||
def send_otp_email(recipient_email, otp):
|
||||
"""Send otp on email with template"""
|
||||
from_email = settings.EMAIL_FROM_ADDRESS
|
||||
recipient_list = [recipient_email]
|
||||
send_templated_mail(
|
||||
@ -26,8 +28,8 @@ def send_otp_email(recipient_email, otp):
|
||||
def custom_response(detail, data=None, response_status=status.HTTP_200_OK):
|
||||
"""Custom response code"""
|
||||
if not data:
|
||||
"""when data is none"""
|
||||
data = None
|
||||
|
||||
return Response({"data": data, "message": detail, "status": "success", "code": response_status})
|
||||
|
||||
|
||||
@ -39,6 +41,7 @@ def custom_error_response(detail, response_status):
|
||||
:return: Json response
|
||||
"""
|
||||
if not detail:
|
||||
"""when details is empty"""
|
||||
detail = {}
|
||||
return Response({"error": detail, "status": "failed", "code": response_status})
|
||||
|
||||
@ -58,16 +61,20 @@ def generate_jwt_token(token_type: str, now_time: int, data: dict = dict):
|
||||
"""
|
||||
if type(data) == type:
|
||||
data = {}
|
||||
"""Update data dictionary"""
|
||||
data.update({
|
||||
'token_type': token_type,
|
||||
'iss': 'your_site_url',
|
||||
'iat': timegm(datetime.utcnow().utctimetuple()),
|
||||
'jti': uuid4().hex
|
||||
})
|
||||
"""Access and Refresh token"""
|
||||
TOKEN_TYPE = ["access", "refresh"]
|
||||
if token_type == TOKEN_TYPE[1]:
|
||||
"""Refresh token"""
|
||||
exp = now_time + settings.SIMPLE_JWT['REFRESH_TOKEN_LIFETIME']
|
||||
else:
|
||||
"""access token"""
|
||||
exp = now_time + settings.SIMPLE_JWT['ACCESS_TOKEN_LIFETIME']
|
||||
|
||||
data.update({
|
||||
@ -84,7 +91,9 @@ def generate_jwt_token(token_type: str, now_time: int, data: dict = dict):
|
||||
def get_token(data: dict = dict):
|
||||
""" create access and refresh token """
|
||||
now_time = datetime.utcnow()
|
||||
"""generate access token"""
|
||||
access = generate_jwt_token('access', now_time, data)
|
||||
"""generate refresh token"""
|
||||
refresh = generate_jwt_token('refresh', now_time, data)
|
||||
|
||||
return {
|
||||
|
@ -1,17 +1,20 @@
|
||||
"""Account view """
|
||||
"""Django import"""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from rest_framework import viewsets, status, views
|
||||
from rest_framework.decorators import action
|
||||
import random
|
||||
import logging
|
||||
from django.utils import timezone
|
||||
import jwt
|
||||
"""App Import"""
|
||||
from guardian.utils import upload_image_to_alibaba
|
||||
from django.contrib.auth import authenticate, login
|
||||
from guardian.models import Guardian
|
||||
from junior.models import Junior
|
||||
from account.models import UserProfile, UserPhoneOtp, UserEmailOtp, DefaultTaskImages
|
||||
from django.contrib.auth.models import User
|
||||
"""Account serializer"""
|
||||
from .serializers import (SuperUserSerializer, GuardianSerializer, JuniorSerializer, EmailVerificationSerializer,
|
||||
ForgotPasswordSerializer, ResetPasswordSerializer, ChangePasswordSerializer,
|
||||
GoogleLoginSerializer, UpdateGuardianImageSerializer, UpdateJuniorProfileImageSerializer,
|
||||
@ -29,12 +32,13 @@ from rest_framework import status
|
||||
from rest_framework.response import Response
|
||||
import requests
|
||||
from django.conf import settings
|
||||
from .utils import get_token
|
||||
from junior.serializers import JuniorProfileSerializer
|
||||
from guardian.serializers import GuardianProfileSerializer
|
||||
|
||||
class GoogleLoginMixin:
|
||||
"""google login mixin"""
|
||||
def google_login(self, request):
|
||||
"""google login function"""
|
||||
access_token = request.data.get('access_token')
|
||||
user_type = request.data.get('user_type')
|
||||
if not access_token:
|
||||
@ -341,7 +345,8 @@ class UserEmailVerification(viewsets.ModelViewSet):
|
||||
guardian_data.save()
|
||||
refresh = RefreshToken.for_user(user_obj)
|
||||
access_token = str(refresh.access_token)
|
||||
return custom_response(SUCCESS_CODE['3011'], {"auth_token":access_token}, response_status=status.HTTP_200_OK)
|
||||
return custom_response(SUCCESS_CODE['3011'], {"auth_token":access_token},
|
||||
response_status=status.HTTP_200_OK)
|
||||
else:
|
||||
return custom_error_response(ERROR_CODE["2008"], response_status=status.HTTP_400_BAD_REQUEST)
|
||||
except Exception as e:
|
||||
|
@ -5,6 +5,7 @@ from django.contrib.auth import get_user_model
|
||||
"""Import Django app"""
|
||||
from base.constants import GENDERS, TASK_STATUS, PENDING, TASK_POINTS
|
||||
from junior.models import Junior
|
||||
"""Add user model"""
|
||||
User = get_user_model()
|
||||
# Create your models here.
|
||||
|
||||
@ -44,16 +45,22 @@ class Guardian(models.Model):
|
||||
return f'{self.user}'
|
||||
|
||||
class JuniorTask(models.Model):
|
||||
"""Guardian model"""
|
||||
"""Junior Task details model"""
|
||||
guardian = models.ForeignKey(Guardian, on_delete=models.CASCADE, related_name='guardian', verbose_name='Guardian')
|
||||
"""task details"""
|
||||
task_name = models.CharField(max_length=100)
|
||||
task_description = models.CharField(max_length=500)
|
||||
"""points of the task"""
|
||||
points = models.IntegerField(default=TASK_POINTS)
|
||||
due_date = models.DateField(auto_now_add=False, null=True, blank=True)
|
||||
"""Images of task"""
|
||||
default_image = models.URLField(null=True, blank=True, default=None)
|
||||
image = models.URLField(null=True, blank=True, default=None)
|
||||
"""associated junior with the task"""
|
||||
junior = models.ForeignKey(Junior, on_delete=models.CASCADE, related_name='junior', verbose_name='Junior')
|
||||
"""task status"""
|
||||
task_status = models.CharField(choices=TASK_STATUS, max_length=15, default=PENDING)
|
||||
"""task stage"""
|
||||
is_active = models.BooleanField(default=True)
|
||||
is_approved = models.BooleanField(default=False)
|
||||
"""Profile created and updated time"""
|
||||
|
@ -114,17 +114,18 @@ class CreateGuardianSerializer(serializers.ModelSerializer):
|
||||
guardian.passcode = validated_data.get('passcode', guardian.passcode)
|
||||
guardian.country_name = validated_data.get('country_name', guardian.country_name)
|
||||
guardian.referral_code_used = validated_data.get('referral_code_used', guardian.referral_code_used)
|
||||
"""Complete profile of the junior if below all data are filled"""
|
||||
complete_profile_field = all([guardian.phone, guardian.gender, guardian.family_name, guardian.country_name,
|
||||
guardian.dob, guardian.country_code, user.first_name, user.last_name])
|
||||
guardian.is_complete_profile = False
|
||||
if complete_profile_field:
|
||||
guardian.is_complete_profile = True
|
||||
image = validated_data.pop('image', None)
|
||||
if image:
|
||||
filename = f"images/{image.name}"
|
||||
image_url = upload_image_to_alibaba(image, filename)
|
||||
guardian.image = image_url
|
||||
"""Complete profile of the junior if below all data are filled"""
|
||||
complete_profile_field = all([guardian.phone, guardian.gender, guardian.family_name, guardian.country_name,
|
||||
guardian.dob, guardian.country_code, user.first_name, user.last_name,
|
||||
user.email, guardian.image])
|
||||
guardian.is_complete_profile = False
|
||||
if complete_profile_field:
|
||||
guardian.is_complete_profile = True
|
||||
guardian.save()
|
||||
return guardian
|
||||
|
||||
@ -137,10 +138,13 @@ class CreateGuardianSerializer(serializers.ModelSerializer):
|
||||
|
||||
|
||||
class TaskSerializer(serializers.ModelSerializer):
|
||||
"""Task serializer"""
|
||||
class Meta(object):
|
||||
"""Meta info"""
|
||||
model = JuniorTask
|
||||
fields = ['task_name','task_description','points', 'due_date', 'junior', 'default_image']
|
||||
def create(self, validated_data):
|
||||
"""create default task image data"""
|
||||
validated_data['guardian'] = Guardian.objects.filter(user=self.context['user']).last()
|
||||
images = self.context['image']
|
||||
validated_data['default_image'] = images
|
||||
@ -173,23 +177,26 @@ class GuardianDetailSerializer(serializers.ModelSerializer):
|
||||
'updated_at']
|
||||
class TaskDetailsSerializer(serializers.ModelSerializer):
|
||||
|
||||
# guardian = GuardianDetailSerializer()
|
||||
junior = JuniorDetailSerializer()
|
||||
class Meta(object):
|
||||
"""Meta info"""
|
||||
model = JuniorTask
|
||||
fields = ['id', 'guardian', 'task_name', 'task_description', 'points', 'due_date','default_image', 'image',
|
||||
'junior', 'task_status', 'is_active', 'created_at','updated_at']
|
||||
|
||||
|
||||
class TopJuniorSerializer(serializers.ModelSerializer):
|
||||
"""Top junior serializer"""
|
||||
junior = JuniorDetailSerializer()
|
||||
position = serializers.SerializerMethodField()
|
||||
|
||||
class Meta:
|
||||
class Meta(object):
|
||||
"""Meta info"""
|
||||
model = JuniorPoints
|
||||
fields = ['id', 'junior', 'total_task_points', 'position', 'created_at', 'updated_at']
|
||||
|
||||
def get_position(self, obj):
|
||||
"""get position of junior"""
|
||||
queryset = self.context['view'].get_queryset()
|
||||
position = list(queryset).index(obj) + 1
|
||||
return position
|
||||
|
@ -1,15 +1,22 @@
|
||||
"""Utiles file of guardian"""
|
||||
"""Django import"""
|
||||
import oss2
|
||||
from django.conf import settings
|
||||
import tempfile
|
||||
|
||||
def upload_image_to_alibaba(image, filename):
|
||||
"""upload image on oss alibaba bucket"""
|
||||
# Save the image object to a temporary file
|
||||
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
||||
"""write image in temporary file"""
|
||||
temp_file.write(image.read())
|
||||
"""auth of bucket"""
|
||||
auth = oss2.Auth(settings.ALIYUN_OSS_ACCESS_KEY_ID, settings.ALIYUN_OSS_ACCESS_KEY_SECRET)
|
||||
"""fetch bucket details"""
|
||||
bucket = oss2.Bucket(auth, settings.ALIYUN_OSS_ENDPOINT, settings.ALIYUN_OSS_BUCKET_NAME)
|
||||
# Upload the temporary file to Alibaba OSS
|
||||
bucket.put_object_from_file(filename, temp_file.name)
|
||||
"""create perfect url for image"""
|
||||
new_filename = filename.replace(' ', '%20')
|
||||
return f"https://{settings.ALIYUN_OSS_BUCKET_NAME}.{settings.ALIYUN_OSS_ENDPOINT}/{new_filename}"
|
||||
|
||||
|
@ -143,7 +143,7 @@ class SearchTaskListAPIView(viewsets.ModelViewSet):
|
||||
class TopJuniorListAPIView(viewsets.ModelViewSet):
|
||||
"""Top juniors list"""
|
||||
serializer_class = TopJuniorSerializer
|
||||
# permission_classes = [IsAuthenticated]
|
||||
permission_classes = [IsAuthenticated]
|
||||
queryset = JuniorPoints.objects.all()
|
||||
|
||||
def get_serializer_context(self):
|
||||
|
@ -89,16 +89,17 @@ class CreateJuniorSerializer(serializers.ModelSerializer):
|
||||
junior.phone = validated_data.get('phone', junior.phone)
|
||||
junior.country_code = validated_data.get('country_code', junior.country_code)
|
||||
junior.referral_code_used = validated_data.get('referral_code_used', junior.referral_code_used)
|
||||
"""Complete profile of the junior if below all data are filled"""
|
||||
complete_profile_field = all([junior.phone, junior.gender, junior.country_name,
|
||||
junior.dob, junior.country_code, user.first_name, user.last_name])
|
||||
junior.is_complete_profile = False
|
||||
if complete_profile_field:
|
||||
junior.is_complete_profile = True
|
||||
if image:
|
||||
filename = f"images/{image.name}"
|
||||
image_url = upload_image_to_alibaba(image, filename)
|
||||
junior.image = image_url
|
||||
"""Complete profile of the junior if below all data are filled"""
|
||||
complete_profile_field = all([junior.phone, junior.gender, junior.country_name, junior.image,
|
||||
junior.dob, junior.country_code, user.first_name, user.last_name,
|
||||
user.email])
|
||||
junior.is_complete_profile = False
|
||||
if complete_profile_field:
|
||||
junior.is_complete_profile = True
|
||||
junior.save()
|
||||
return junior
|
||||
|
||||
|
@ -1,7 +1,9 @@
|
||||
#!/usr/bin/env python
|
||||
"""Django's command-line utility for administrative tasks."""
|
||||
"""Django import"""
|
||||
"""Import OS module"""
|
||||
import os
|
||||
"""Import sys module"""
|
||||
import sys
|
||||
|
||||
|
||||
@ -18,6 +20,7 @@ def main():
|
||||
"available on your PYTHONPATH environment variable? Did you "
|
||||
"forget to activate a virtual environment?"
|
||||
) from exc
|
||||
"""execute command line function"""
|
||||
execute_from_command_line(sys.argv)
|
||||
|
||||
|
||||
|
@ -94,7 +94,7 @@ REST_FRAMEWORK = {
|
||||
'rest_framework.authentication.BasicAuthentication',
|
||||
'rest_framework_simplejwt.authentication.JWTAuthentication',],
|
||||
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
|
||||
'PAGE_SIZE': 5, # Set the default pagination size
|
||||
'PAGE_SIZE': 5,
|
||||
}
|
||||
SIMPLE_JWT = {
|
||||
'ACCESS_TOKEN_LIFETIME': timedelta(minutes=50),
|
||||
@ -201,22 +201,15 @@ https://docs.djangoproject.com/en/3.0/howto/static-files/"""
|
||||
GOOGLE_CLIENT_ID = os.getenv('GOOGLE_CLIENT_ID')
|
||||
GOOGLE_CLIENT_SECRET = os.getenv('GOOGLE_CLIENT_SECRET')
|
||||
|
||||
# EMAIL_BACKEND = os.getenv('EMAIL_BACKEND')
|
||||
# EMAIL_HOST = os.getenv('EMAIL_HOST')
|
||||
# EMAIL_PORT = os.getenv('EMAIL_PORT')
|
||||
# EMAIL_USE_TLS = os.getenv('EMAIL_USE_TLS')
|
||||
# EMAIL_HOST_USER = os.getenv('EMAIL_HOST_USER') # Replace with your Gmail email address
|
||||
# EMAIL_HOST_PASSWORD = os.getenv('EMAIL_HOST_PASSWORD') # Replace with your Gmail email password or App password
|
||||
# EMAIL_FROM_ADDRESS = os.getenv('EMAIL_FROM_ADDRESS')
|
||||
|
||||
EMAIL_BACKEND="django.core.mail.backends.smtp.EmailBackend"
|
||||
EMAIL_HOST="smtp.sendgrid.net"
|
||||
EMAIL_PORT="587"
|
||||
EMAIL_USE_TLS="True"
|
||||
EMAIL_HOST_USER="apikey" # Replace with your Gmail email address
|
||||
EMAIL_HOST_USER="apikey"
|
||||
EMAIL_HOST_PASSWORD="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"
|
||||
EMAIL_FROM_ADDRESS="support@zodbank.com"
|
||||
# EMAIL_FROM_ADDRESS="zodbank@yopmail.com"
|
||||
|
||||
|
||||
ALIYUN_OSS_ACCESS_KEY_ID = os.getenv('ALIYUN_OSS_ACCESS_KEY_ID')
|
||||
ALIYUN_OSS_ACCESS_KEY_SECRET = os.getenv('ALIYUN_OSS_ACCESS_KEY_SECRET')
|
||||
|
Reference in New Issue
Block a user