This the backend source code of tech cafe website. used django rest framework and postgres database 
from django.contrib import admin
from .models import User, Languages, Frameworks
from django.contrib.auth.admin import UserAdmin


class UserAdminConfig(UserAdmin):
    model = User
    search_fields = ("phone", "first_name", "last_name")
    list_filter = (
        "is_active",
        "is_staff",
        "is_superuser",
        "created_date",
        "is_ban",
    )
    ordering = ("-created_date",)
    list_display = (
        "phone",
        "first_name",
        "last_name",
        "job_field",
        "is_superuser",
        "is_staff",
        "is_ban",
        "is_active",
        "created_date",
    )

    fieldsets = (
        (
            "Account Data",
            {
                "fields": (
                    "phone",
                    "email",
                    "first_name",
                    "last_name",
                    "job_field",
                    "avatar",
                ),
            },
        ),
        ("languages &frameworks ", {"fields": ("languages", "frameworks")}),
        ("Event Cancellation Count", {"fields": ("is_ban",)}),
        ("Permissions", {"fields": ("is_staff", "is_active", "is_superuser")}),
        (
            "Group Permissions",
            {
                "fields": (
                    "groups",
                    "user_permissions",
                ),
            },
        ),
        ("Important dates", {"fields": ("last_login",)}),
    )
    add_fieldsets = (
        (
            None,
            {
                "classes": ("wide",),
                "fields": (
                    "email",
                    "phone",
                    "first_name",
                    "last_name",
                    "job_field",
                    "frameworks",
                    "languages",
                    "avatar",
                    "is_ban",
                    "password1",
                    "password2",
                    "is_active",
                    "is_staff",
                    "is_superuser",
                ),
            },
        ),
    )


admin.site.site_header = "Tech Cafe Admin Panel"
admin.site.register(User, UserAdminConfig)
admin.site.register(Languages)
admin.site.register(Frameworks)

core/accounts/admin.py
H-VICTOOR/TechCafe-Backend
from rest_framework import serializers
from accounts.models import User, Languages, Frameworks
from django.contrib.auth.password_validation import validate_password
from django.core import exceptions
from django.core.cache import cache
from validators.fieldvalidators import FieldValidators
from utils.constants import Errors
from django.shortcuts import get_object_or_404
from django.core.cache import cache
from django.conf import settings
from accounts.tasks import send_otp
import random
from rest_framework import status

# Serializer for user registration with password checkup
class RegisterSerializer(serializers.ModelSerializer):
    """Registration serializer with password checkup"""

    password = serializers.CharField(max_length=68, min_length=6, write_only=True)
    password1 = serializers.CharField(max_length=68, min_length=6, write_only=True)
    otp_code = serializers.CharField(
        max_length=4,
        validators=[FieldValidators.xss_validator, FieldValidators.otp_validator],
        write_only=True,
    )

    class Meta:
        model = User
        fields = [
            "id",
            "phone",
            "otp_code",
            "first_name",
            "last_name",
            "password",
            "password1",
        ]
        required_fields = ["phone", "otp_code", "first_name", "last_name"]

    def validate(self, attrs):
        # Check if passwords match
        if attrs.get("password") != attrs.get("password1"):
            raise serializers.ValidationError(Errors.PASSWORD_MISMATCHED)

        # Validate OTP code
        if cache.get(f"register:{attrs.get('phone')}") != None:
            if attrs.get("otp_code") == cache.get(f"register:{attrs.get('phone')}"):
                cache.delete(f"register:{attrs.get('phone')}")
                attrs.pop("otp_code", None)
            else:
                raise serializers.ValidationError(Errors.INVALID_OTP_CODE)
        else:
            raise serializers.ValidationError(Errors.INVALID_OTP_CODE)
        return super().validate(attrs)

    def create(self, validated_data):
        if validated_data.get("email") == "":
            validated_data["email"] = None
        validated_data.pop("password1")
        return User.objects.create_user(**validated_data)


# Serializer for managing user profile info
class ProfileSerializer(serializers.ModelSerializer):
    """Profile serializer to manage extra user info"""

    languages = serializers.SlugRelatedField(
        many=True, slug_field="name", queryset=Languages.objects.all()
    )
    frameworks = serializers.SlugRelatedField(
        many=True, slug_field="name", queryset=Frameworks.objects.all()
    )
    otp_code = serializers.CharField(
        max_length=4,
        validators=[FieldValidators.xss_validator, FieldValidators.otp_validator],
        write_only=True,
        required=False,
    )

    class Meta:
        model = User
        fields = [
            "id",
            "phone",
            "otp_code",
            "first_name",
            "last_name",
            "email",
            "avatar",
            "job_field",
            "languages",
            "frameworks",
        ]
        read_only_fields = [
            "id",
        ]

    def validate(self, attrs):
        if attrs.get("phone") != None:
            try:
                db_phone = User.objects.get(phone=attrs.get("phone"))
            except User.DoesNotExist:
                if attrs.get("otp_code") != None:
                    if cache.get(f"register:{attrs.get('phone')}") != None:
                        if attrs.get("otp_code") == cache.get(
                            f"register:{attrs.get('phone')}"
                        ):
                            cache.delete(f"register:{attrs.get('phone')}")
                            attrs.pop("otp_code", None)
                        else:
                            raise serializers.ValidationError(Errors.INVALID_OTP_CODE)
                    else:
                        raise serializers.ValidationError(Errors.INVALID_OTP_CODE)
                else:
                    raise serializers.ValidationError(Errors.OTP_BLANK)
            else:
                if attrs.get("phone") == db_phone.phone:
                    pass

        return super().validate(attrs)


# Serializer for changing user password
class ChangePasswordSerializer(serializers.Serializer):
    """Change password serializer"""

    old_password = serializers.CharField(required=True)
    new_password = serializers.CharField(required=True)
    new_password1 = serializers.CharField(required=True)

    def validate(self, attrs):
        if attrs.get("new_password") != attrs.get("new_password1"):
            raise serializers.ValidationError(Errors.PASSWORD_MISMATCHED)
        try:
            validate_password(attrs.get("new_password"))
        except exceptions.ValidationError as e:
            raise serializers.ValidationError({"new_password": list(e.messages)})

        return super().validate(attrs)


# Serializer for resetting user password using OTP
class ResetPasswordSerializer(serializers.Serializer):
    """Reset password serializer"""

    phone = serializers.CharField(
        max_length=11,
        validators=[
            FieldValidators.iranian_phone_validator,
            FieldValidators.xss_validator,
        ],
    )
    otp_code = serializers.CharField(
        max_length=4,
        validators=[FieldValidators.xss_validator, FieldValidators.otp_validator],
    )
    new_password = serializers.CharField(required=True)
    new_password1 = serializers.CharField(required=True)

    def validate(self, attrs):
        if attrs.get("new_password") != attrs["new_password1"]:
            raise serializers.ValidationError(Errors.PASSWORD_MISMATCHED)
        try:
            validate_password(attrs.get("new_password"))
        except exceptions.ValidationError as e:
            raise serializers.ValidationError({"new_password": list(e.messages)})
        return super().validate(attrs)


# Serializer for sending OTP
class OTPSendSerializer(serializers.Serializer):
    """Serializer for Send OTP"""

    OTP_TYPE_CHOICES = (
        ("register_otp", "Register OTP"),
        ("reset_password_otp", "Reset Password OTP"),
    )
    phone = serializers.CharField(
        max_length=11,
        validators=[
            FieldValidators.iranian_phone_validator,
            FieldValidators.xss_validator,
        ],
    )
    otp_type = serializers.ChoiceField(choices=OTP_TYPE_CHOICES, allow_null=False)

    class Meta:
        fields = [
            "phone",
        ]

    def set_cache(self):
        phone = self.validated_data["phone"]
        otp_type = self.validated_data["otp_type"]
        if otp_type == "register_otp":
            user_obj = User.objects.filter(phone=phone).exists()
            if not user_obj:
                if cache.get(f"register:{phone}") == None:
                    generate_otp = random.randint(1000, 9999)
                    cache.set(
                        f"register:{phone}", f"{generate_otp}", settings.OTP_CODE_TIME
                    )
                    if send_otp.delay(phone, generate_otp):
                        return {"detail": "ok"}, status.HTTP_200_OK
                    else:
                        return (
                            Errors.SMS_PANEL,
                            status.HTTP_502_BAD_GATEWAY,
                        )
                else:
                    return Errors.OTP_SPAM, status.HTTP_406_NOT_ACCEPTABLE
            else:
                return {
                    "detail": "The User Already Registered"
                }, status.HTTP_406_NOT_ACCEPTABLE
        elif otp_type == "reset_password_otp":
            get_object_or_404(User, phone=phone)
            if cache.get(f"reset_password:{phone}") == None:
                generate_otp = random.randint(1000, 9999)
                cache.set(
                    f"reset_password:{phone}",
                    f"{generate_otp}",
                    settings.OTP_CODE_TIME,
                )
                if send_otp.delay(phone, generate_otp):
                    return {"detail": "ok"}, status.HTTP_200_OK
                else:
                    return (
                        Errors.SMS_PANEL,
                        status.HTTP_502_BAD_GATEWAY,
                    )
            else:
                return Errors.OTP_SPAM, status.HTTP_406_NOT_ACCEPTABLE

core/accounts/api/serializers.py
H-VICTOOR/TechCafe-Backend
# from rest_framework.authtoken.views import ObtainAuthToken
from django.urls import path
from .views import (
    RegisterApiView,
    ProfileApiView,
    ChangePasswordAPIView,
    ResetPasswordAPIView,
    OTPSendAPIView,
)
from rest_framework_simplejwt.views import (
    TokenObtainPairView,
    TokenRefreshView,
    TokenVerifyView,
)


urlpatterns = [
    path("register/", RegisterApiView.as_view(), name="register"),
    # path("token/login", ObtainAuthToken.as_view(), name="token_obtain"),
    path("jwt/create/", TokenObtainPairView.as_view(), name="jwt_obtain_pair"),
    path("jwt/refresh/", TokenRefreshView.as_view(), name="jwt_refresh"),
    path("jwt/verify/", TokenVerifyView.as_view(), name="jwt_verify"),
    path("user/profile/", ProfileApiView.as_view(), name="profile"),
    path(
        "user/change-password/", ChangePasswordAPIView.as_view(), name="change_password"
    ),
    path("reset-password/", ResetPasswordAPIView.as_view(), name="change_password"),
    path("otp/send/", OTPSendAPIView.as_view(), name="otp_register_send"),
]

core/accounts/api/urls.py
H-VICTOOR/TechCafe-Backend
from rest_framework.response import Response
from rest_framework import status
from rest_framework import generics
from rest_framework.permissions import IsAuthenticated
from django.shortcuts import get_object_or_404
from .serializers import (
    ProfileSerializer,
    RegisterSerializer,
    ChangePasswordSerializer,
    ResetPasswordSerializer,
    OTPSendSerializer,
)
from accounts.models import User
import random
from django.core.cache import cache
from accounts.tasks import send_otp
from utils.constants import Errors
from django.conf import settings

# View to handle user registration.
class RegisterApiView(generics.CreateAPIView):
    """
    When a user registers, they send their registration data to this view. It uses the RegisterSerializer to validate and process the input data, creating a new user account in the process.
    """

    serializer_class = RegisterSerializer
    queryset = User.objects.all()

# View to retrieve and update user profiles.
class ProfileApiView(generics.RetrieveUpdateAPIView):

    """
    Authenticated users can access this view to retrieve or modify their profile details. It uses the ProfileSerializer for serialization and requires authentication (IsAuthenticated) for access.
    """
    queryset = User.objects.all()
    serializer_class = ProfileSerializer
    permission_classes = [IsAuthenticated]

    # For not sending pk in URL:
    def get_object(self):
        queryset = self.get_queryset()
        obj = get_object_or_404(queryset, phone=self.request.user)
        return obj

# View to change a user's password.
class ChangePasswordAPIView(generics.GenericAPIView):
    """
    Authenticated users can use this view to update their passwords. It checks the old password for validation and updates the user's password with the new one. The ChangePasswordSerializer handles input validation.
    """
    queryset = User.objects.all()
    permission_classes = [IsAuthenticated]
    serializer_class = ChangePasswordSerializer

    def get_object(self):
        obj = self.request.user
        return obj

    def put(self, request, *args, **kwargs):
        self.object = self.get_object()
        serializer = self.get_serializer(data=request.data)
        if serializer.is_valid():
            # Check old password
            if not self.object.check_password(serializer.data.get("old_password")):
                return Response(
                    {"old_password": ["Wrong password."]},
                    status=status.HTTP_400_BAD_REQUEST,
                )
            # set_password also hashes the password that the user will get
            self.object.set_password(serializer.data.get("new_password"))
            self.object.save()
            return Response(
                {"detail": "password changed successfully"}, status=status.HTTP_200_OK
            )

        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

# View to reset a user's password using OTP.
class ResetPasswordAPIView(generics.CreateAPIView):
    """
    Users who forget their passwords can request a reset through this view. It validates the OTP code and updates the password if the code is correct. Caching is used to store and validate OTP codes.
    """
    model = User
    serializer_class = ResetPasswordSerializer

    def create(self, request):
        serializer = self.serializer_class(data=request.data)
        if serializer.is_valid():
            if cache.get(f"reset_password:{serializer.data['phone']}") != None:
                if queryset := get_object_or_404(User, phone=serializer.data["phone"]):
                    if serializer.data["otp_code"] == cache.get(
                        f"reset_password:{serializer.data['phone']}"
                    ):
                        queryset.set_password(serializer.data.get("new_password"))
                        queryset.save()
                        cache.delete(f"reset_password:{serializer.data['phone']}")
                        return Response(
                            {"detail": "password successfully changed"},
                            status=status.HTTP_200_OK,
                        )
                    else:
                        return Response(
                            Errors.INVALID_OTP_CODE,
                            status=status.HTTP_406_NOT_ACCEPTABLE,
                        )
                else:
                    return Response(
                        {"detail": "user not found"}, status=status.HTTP_404_NOT_FOUND
                    )
            else:
                return Response(
                    Errors.INVALID_OTP_CODE,
                    status=status.HTTP_406_NOT_ACCEPTABLE,
                )
        else:
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

# View to send OTP for password reset.
class OTPSendAPIView(generics.CreateAPIView):
    serializer_class = OTPSendSerializer
    """
    Users can request OTP codes for actions like password reset by sending a request to this view. It generates and sends the OTP code to the user's phone number, utilizing the OTPSendSerializer. The code is stored in the cache for later verification.
    """

    def create(self, request):
        serializer = self.serializer_class(data=request.data)
        serializer.is_valid(raise_exception=True)
        result, status = serializer.set_cache()
        return Response(result, status)

core/accounts/api/views.py
H-VICTOOR/TechCafe-Backend
from django.dispatch import receiver
from django.db.models.signals import post_save
from django.contrib.auth.base_user import BaseUserManager
from django.utils.translation import gettext_lazy as _
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin
from django.db import models
from accounts.tasks import send_welcome
from validators.fieldvalidators import FieldValidators
from validators.volumevalidator import VolumeValidator

# Define a custom user manager for the User model.
class UserManager(BaseUserManager):
    """
    Custom user model manager where phone is the unique identifier
    for authentication instead of usernames.
    """

    def create_user(self, phone, password, **extra_fields):
        """
        Create and save a User with the given phone and password.
        """
        if not phone:
            raise ValueError(_("The phone must be set"))
        extra_fields.setdefault("is_active", True)
        user = self.model(phone=phone, **extra_fields)
        user.set_password(password)
        user.save()
        return user

    def create_superuser(self, phone, password, **extra_fields):
        """
        Create and save a SuperUser with the given phone and password.
        """
        extra_fields.setdefault("is_staff", True)
        extra_fields.setdefault("is_superuser", True)
        extra_fields.setdefault("is_active", True)
        extra_fields.setdefault("is_ban", False)

        if extra_fields.get("is_staff") is not True:
            raise ValueError(_("Superuser must have is_staff=True."))
        if extra_fields.get("is_superuser") is not True:
            raise ValueError(_("Superuser must have is_superuser=True."))
        return self.create_user(phone, password, **extra_fields)

# Define the custom User model.
class User(AbstractBaseUser, PermissionsMixin):
    phone = models.CharField(
        _("Phone"),
        max_length=11,
        unique=True,
        validators=[
            FieldValidators.iranian_phone_validator,
            FieldValidators.xss_validator,
        ],
        blank=False,
        null=False,
    )
    email = models.EmailField(
        validators=[FieldValidators.xss_validator], unique=True, blank=True, null=True
    )
    first_name = models.CharField(
        max_length=255,
        validators=[FieldValidators.xss_validator],
        blank=False,
        null=False,
    )
    last_name = models.CharField(
        max_length=255,
        validators=[FieldValidators.xss_validator],
        blank=False,
        null=False,
    )
    job_field = models.CharField(
        max_length=100,
        validators=[FieldValidators.xss_validator],
        blank=True,
        null=True,
    )
    avatar = models.ImageField(
        upload_to="avatars",
        validators=[VolumeValidator.validate_avatar_size],
        blank=True,
        null=True,
    )
    languages = models.ManyToManyField("languages", blank=True)
    frameworks = models.ManyToManyField("frameworks", blank=True)
    is_staff = models.BooleanField(default=False)
    is_active = models.BooleanField(default=False)
    is_ban = models.BooleanField(default=False)
    created_date = models.DateTimeField(auto_now_add=True)
    updated_date = models.DateTimeField(auto_now=True)

    USERNAME_FIELD = "phone"
    REQUIRED_FIELDS = []

    objects = UserManager()

    def __str__(self):
        return self.phone

    @property
    def fullname(self):
        return f"{self.first_name} {self.last_name}"

# Define models for Languages and Frameworks.
class Languages(models.Model):
    name = models.CharField(max_length=250)

    def __str__(self):
        return self.name

class Frameworks(models.Model):
    name = models.CharField(max_length=250)

    def __str__(self):
        return self.name

# Signal handler to create a user profile and send a welcome message on user creation.
@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    if created:
        send_welcome.delay(instance.phone, instance.first_name)

core/accounts/models.py
H-VICTOOR/TechCafe-Backend
from celery import shared_task
from adapter.melipayamak import Api
from decouple import config

username = config("MELIPAYAMAK_USERNAME", default="test")
password = config("MELIPAYAMAK_PASSWORD", default="test")


@shared_task
def send_otp(phone, code):
    api = Api(username, password)
    sms_rest = api.sms()
    to = phone
    text = [
        f"{code}",
    ]
    bodyId = 115131

    response = sms_rest.send_by_base_number(text, to, bodyId)
    if len(response["Value"]) >= 15:
        return True
    else:
        return False


@shared_task
def send_welcome(phone, first_name):
    api = Api(username, password)
    sms_rest = api.sms()
    to = phone
    text = [
        f"{first_name}",
    ]
    bodyId = 115128

    response = sms_rest.send_by_base_number(text, to, bodyId)
    if len(response["Value"]) >= 15:
        return True
    else:
        return False

core/accounts/tasks.py
H-VICTOOR/TechCafe-Backend
from rest_framework.test import APIClient
from django.urls import reverse
import pytest


class TestPostApi:
    def test_get_post_response_200_status(self):
        client = APIClient()
        url = reverse("profile")
        response = client.get(url)
        assert response.status_code == 200

    def test_create_post_response_401_status(self):
        url = reverse()
        data = {
            "phone": "09121234567",
            "first_name": "test_first_name",
            "last_name": "test_last_name",
            "language": "Python",
            "framework": "django",
        }
        response = self.client.post(url, data)
        assert response.status_code == 401

core/accounts/tests/test_api.py
H-VICTOOR/TechCafe-Backend
from django.contrib.auth import get_user_model
from django.test import TestCase


class UsersManagersTests(TestCase):
    def test_create_user(self):
        User = get_user_model()
        user = User.objects.create_user(phone="00000000000", password="a/@1234567")
        self.assertEqual(user.phone, "00000000000")
        self.assertFalse(user.is_active)
        self.assertFalse(user.is_staff)
        self.assertFalse(user.is_superuser)
        try:
            # username is None for the AbstractUser option
            # username does not exist for the AbstractBaseUser option
            self.assertIsNone(user.username)
        except AttributeError:
            pass
        with self.assertRaises(TypeError):
            User.objects.create_user()
        with self.assertRaises(TypeError):
            User.objects.create_user(phone="")
        with self.assertRaises(ValueError):
            User.objects.create_user(phone="", password="a/@1234567")

    def test_create_superuser(self):
        User = get_user_model()
        admin_user = User.objects.create_superuser(
            phone="00000000000", password="a/@1234567"
        )
        self.assertEqual(admin_user.phone, "00000000000")
        self.assertTrue(admin_user.is_active)
        self.assertTrue(admin_user.is_staff)
        self.assertTrue(admin_user.is_superuser)
        try:
            # username is None for the AbstractUser option
            # username does not exist for the AbstractBaseUser option
            self.assertIsNone(admin_user.username)
        except AttributeError:
            pass
        with self.assertRaises(ValueError):
            User.objects.create_superuser(
                phone="00000000000", password="foo", is_superuser=False
            )

core/accounts/tests/test_model.py
H-VICTOOR/TechCafe-Backend
from django.urls import path, include

app_name = "accounts"

urlpatterns = [
    # template base authentication
    # path('', include('django.contrib.auth.urls')),
    # api based authentication
    path("api/v1/", include("accounts.api.urls"))
]

core/accounts/urls.py
H-VICTOOR/TechCafe-Backend
import os

from celery import Celery
from django.conf import settings

# Set the default Django settings module for the 'celery' program.
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")

app = Celery("core")

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object(settings, namespace="CELERY")


# Load task modules from all registered Django apps.
app.autodiscover_tasks()

core/core/celery.py
H-VICTOOR/TechCafe-Backend
"""
Django settings for core project.

Generated by 'django-admin startproject' using Django 4.1.3.

For more information on this file, see
https://docs.djangoproject.com/en/4.1/topics/settings/

For the full list of settings and their values, see
https://docs.djangoproject.com/en/4.1/ref/settings/
"""

from pathlib import Path
from datetime import timedelta
from decouple import config

# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent


# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/4.1/howto/deployment/checklist/

# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = config("SECRET_KEY", default="test")

# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = config("DEBUG", cast=bool, default=False)

ALLOWED_HOSTS = ["api.techcafe.ir"]

CSRF_TRUSTED_ORIGINS = [
    "https://*.127.0.0.1",
    "https://api.techcafe.ir",
    "https://techcafe.ir",
]

if DEBUG : 
    CORS_ALLOW_ALL_ORIGINS = True
else:
    CORS_ALLOWED_ORIGINS = [
        "https://techcafe.ir",
        "http://localhost:8080",
        "https://api.techcafe.ir",
        "http://api.techcafe.ir",
        "https://www.techcafe.ir",
    ]

# Application definition

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "rest_framework",
    "rest_framework.authtoken",
    "rest_framework_simplejwt",
    "drf_yasg",
    "corsheaders",
    "accounts.apps.AccountsConfig",
    "gathering.apps.GatheringConfig",
    "django_filters",
    "silk",
    "azbankgateways",
]

MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django.middleware.clickjacking.XFrameOptionsMiddleware",
    "silk.middleware.SilkyMiddleware",
    "corsheaders.middleware.CorsMiddleware",
    "django.middleware.common.CommonMiddleware",
]

ROOT_URLCONF = "core.urls"

TEMPLATES = [
    {
        "BACKEND": "django.template.backends.django.DjangoTemplates",
        "DIRS": [],
        "APP_DIRS": True,
        "OPTIONS": {
            "context_processors": [
                "django.template.context_processors.debug",
                "django.template.context_processors.request",
                "django.contrib.auth.context_processors.auth",
                "django.contrib.messages.context_processors.messages",
            ],
        },
    },
]

WSGI_APPLICATION = "core.wsgi.application"


# Database
# https://docs.djangoproject.com/en/4.1/ref/settings/#databases

if DEBUG :
    DATABASES = {
        "default": {
            "ENGINE": "django.db.backends.sqlite3",
            "NAME": BASE_DIR / "db.sqlite3",
        }
    }
else:
    DATABASES = {
        "default": {
            "ENGINE": config("DB_ENGINE", default="django.db.backends.postgresql"),
            "NAME": config("DB_NAME", default="postgres"),
            "USER": config("DB_USER", default="postgres"),
            "PASSWORD": config("DB_PASS", default="postgres"),
            "HOST": config("DB_HOST", default="db"),
            "PORT": config("DB_PORT", cast=int, default=5432),
        }
    }

# Password validation
# https://docs.djangoproject.com/en/4.1/ref/settings/#auth-password-validators

AUTH_PASSWORD_VALIDATORS = [
    {
        "NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator",
    },
    {
        "NAME": "django.contrib.auth.password_validation.MinimumLengthValidator",
    },
    {
        "NAME": "django.contrib.auth.password_validation.CommonPasswordValidator",
    },
    {
        "NAME": "django.contrib.auth.password_validation.NumericPasswordValidator",
    },
]


# Internationalization
# https://docs.djangoproject.com/en/4.1/topics/i18n/

LANGUAGE_CODE = "en-us"

# TIME_ZONE = "UTC"
TIME_ZONE = "Asia/Tehran"
USE_I18N = True

USE_TZ = True


# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/4.1/howto/static-files/

STATIC_ROOT = BASE_DIR / "static"
STATIC_URL = "/static/"
MEDIA_ROOT = BASE_DIR / "media"
MEDIA_URL = "/media/"

STATICFILES_DIRS = [
    (BASE_DIR / "staticfiles"),
]

# Default primary key field type
# https://docs.djangoproject.com/en/4.1/ref/settings/#default-auto-field

DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"


AUTH_USER_MODEL = "accounts.User"

REST_FRAMEWORK = {
    "DEFAULT_PERMISSION_CLASSES": [
        "rest_framework.permissions.AllowAny",
    ],
    "DEFAULT_AUTHENTICATION_CLASSES": [
        "rest_framework_simplejwt.authentication.JWTAuthentication",
        "rest_framework.authentication.TokenAuthentication",
        "rest_framework.authentication.BasicAuthentication",
        "rest_framework.authentication.SessionAuthentication",
    ],
    "DEFAULT_THROTTLE_CLASSES": [
        "rest_framework.throttling.AnonRateThrottle",
        "rest_framework.throttling.UserRateThrottle",
    ],
    "DEFAULT_THROTTLE_RATES": {"anon": "20/minute", "user": "40/minute"},
}
if not DEBUG:
    REST_FRAMEWORK.update(
        {"DEFAULT_RENDERER_CLASSES": ("rest_framework.renderers.JSONRenderer",)}
    )

# time for clear cache (OTP SYSTEM )
OTP_CODE_TIME = 180


# json web token configs
SIMPLE_JWT = {
    "ACCESS_TOKEN_LIFETIME": timedelta(days=7),
    "REFRESH_TOKEN_LIFETIME": timedelta(days=14),
    "ROTATE_REFRESH_TOKENS": False,
    "BLACKLIST_AFTER_ROTATION": False,
    "UPDATE_LAST_LOGIN": True,
}

# Caching configuration1
CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": "redis://redis:6379",
        "OPTIONS": {"CLIENT_CLASS": "django_redis.client.DefaultClient"},
    }
}

# celery configuration

CELERY_BROKER_URL = "redis://redis:6379/1"


AZ_IRANIAN_BANK_GATEWAYS = {
    "GATEWAYS": {
        "BMI": {
            "MERCHANT_CODE": "<YOUR MERCHANT CODE>",
            "TERMINAL_CODE": "<YOUR TERMINAL CODE>",
            "SECRET_KEY": "<YOUR SECRET CODE>",
        },
        "SEP": {
            "MERCHANT_CODE": "<YOUR MERCHANT CODE>",
            "TERMINAL_CODE": "<YOUR TERMINAL CODE>",
        },
        "ZARINPAL": {
            "MERCHANT_CODE": "<YOUR MERCHANT CODE>",
            "SANDBOX": 1,  # 0 disable, 1 active
        },
        "IDPAY": {
            "MERCHANT_CODE": "<YOUR MERCHANT CODE>",
            "METHOD": "POST",  # GET or POST
            "X_SANDBOX": 0,  # 0 disable, 1 active
        },
        "ZIBAL": {
            "MERCHANT_CODE": "<YOUR MERCHANT CODE>",
        },
        "BAHAMTA": {
            "MERCHANT_CODE": "<YOUR MERCHANT CODE>",
        },
        "MELLAT": {
            "TERMINAL_CODE": "<YOUR TERMINAL CODE>",
            "USERNAME": "<YOUR USERNAME>",
            "PASSWORD": "<YOUR PASSWORD>",
        },
        "PAYV1": {
            "MERCHANT_CODE": "<YOUR MERCHANT CODE>",
            "X_SANDBOX": 0,  # 0 disable, 1 active
        },
    },
    "IS_SAMPLE_FORM_ENABLE": False,  # اختیاری و پیش فرض غیر فعال است
    "DEFAULT": "ZARINPAL",
    "CURRENCY": "IRR",  # اختیاری
    "BANK_PRIORITIES": [
        "ZARINPAL",
    ],  # اختیاری
    "IS_SAFE_GET_GATEWAY_PAYMENT": True,  # اختیاری، بهتر است True بزارید.
    "CUSTOM_APP": None,  # اختیاری
}

core/core/settings.py
H-VICTOOR/TechCafe-Backend
"""core URL Configuration
The `urlpatterns` list routes URLs to views. For more information please see:
    https://docs.djangoproject.com/en/3.2/topics/http/urls/
Examples:
Function views
    1. Add an import:  from my_app import views
    2. Add a URL to urlpatterns:  path('', views.home, name='home')
Class-based views
    1. Add an import:  from other_app.views import Home
    2. Add a URL to urlpatterns:  path('', Home.as_view(), name='home')
Including another URLconf
    1. Import the include() function: from django.urls import include, path
    2. Add a URL to urlpatterns:  path('blog/', include('blog.urls'))
"""
from django.contrib import admin
from django.urls import path, include
from rest_framework import permissions
from django.conf import settings
from django.conf.urls.static import static
from drf_yasg.views import get_schema_view
from drf_yasg import openapi
from decouple import config
from core.views import admin_meme
from azbankgateways.urls import az_bank_gateways_urls

schema_view = get_schema_view(
    openapi.Info(
        title="Tech Cafe Rest API",
        default_version="v 1.0.0",
        description="Tech Cafe Rest API Backend Service",
        # terms_of_service="https://www.google.com/policies/terms/",
        contact=openapi.Contact(email="hamidfarzin1382@gmail.com"),
        license=openapi.License(name="MIT License"),
    ),
    public=True,
    permission_classes=(permissions.IsAdminUser,),
)

urlpatterns = [
    # admin panel
    path(f"{config('ADMIN_URL', default='test')}/", admin.site.urls),
    # admin panel meme
    path("admin/", admin_meme),
    # accounts app
    path("accounts/", include("accounts.urls")),
    # gathering app
    path("gathering/", include("gathering.urls")),
    # api authentication
    path("api-auth/", include("rest_framework.urls", namespace="rest_framework")),
    # api docs
    path(
        "swagger<format>/", schema_view.without_ui(cache_timeout=0), name="schema-json"
    ),
    path(
        "swagger/",
        schema_view.with_ui("swagger", cache_timeout=0),
        name="schema-swagger-ui",
    ),
    path("redoc/", schema_view.with_ui("redoc", cache_timeout=0), name="schema-redoc"),
    # silk
    path(
        f"{config('SILK_URL', default='silk')}/", include("silk.urls", namespace="silk")
    ),
    path("bankgateways/", az_bank_gateways_urls()),
]


if settings.DEBUG:
    urlpatterns += static(settings.STATIC_URL, document_root=settings.STATIC_ROOT)
    urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

core/core/urls.py
H-VICTOOR/TechCafe-Backend
from django.http import HttpResponse


def admin_meme(requests):
    return HttpResponse("<center> <h1> hihi, try to think smarter! </h1> </center>")

core/core/views.py
H-VICTOOR/TechCafe-Backend
from django.contrib import admin
from gathering.models import Gathering, Photo, GatheringUser, Discount
import os

# Register your models here.


class GatheringAdmin(admin.ModelAdmin):
    search_fields = ["title", "presenter"]
    list_filter = ["title", "date", "is_held", "is_online", "is_occupied"]
    ordering = [
        "-date",
    ]
    list_display = (
        "title",
        "price",
        "date",
        "max_seats",
        "is_held",
        "is_online",
        "is_occupied",
    )


class GatheringUserAdmin(admin.ModelAdmin):
    search_fields = ["gathering__id", "gathering__title", "user__phone", "uuid"]
    list_filter = ["gathering", "check_in", "discount"]
    list_display = ("user", "user_full_name", "gathering", "uuid", "check_in")
    readonly_fields = ["uuid", "created_date", "updated_date"]

    def user_full_name(self, obj):
        return f"{obj.user.first_name} {obj.user.last_name}"


class PhotoAdmin(admin.ModelAdmin):
    search_fields = ["gathering__id", "image", "user"]
    list_display = [
        "image",
    ]


class DiscountAdmin(admin.ModelAdmin):
    search_fields = ["code", "gathering__title"]
    list_filter = ["status"]
    list_display = ("code", "gathering", "discount_percentage", "status")

    def user_full_name(self, obj):
        return f"{obj.user.first_name} {obj.user.last_name}"


admin.site.register(Gathering, GatheringAdmin)
admin.site.register(Photo, PhotoAdmin)
admin.site.register(GatheringUser, GatheringUserAdmin)
admin.site.register(Discount, DiscountAdmin)

core/gathering/admin.py
H-VICTOOR/TechCafe-Backend
from rest_framework import serializers
from gathering.models import Gathering, GatheringUser, Photo, Discount
from accounts.models import User
from utils.constants import Errors
from django.urls import reverse
from gathering.services.bank_gateway import Gateway
from rest_framework import status
from django.utils import timezone
import datetime

# Serializer for gathering details (read-only)
class GatheringSerializer(serializers.ModelSerializer):
    absolute_url = serializers.SerializerMethodField(method_name="get_abs_url")
    snippet = serializers.ReadOnlyField(source="get_snippet")
    presenter = serializers.SlugRelatedField(
        many=True,
        read_only=True,
        slug_field=("fullname"),
    )

    class Meta:
        model = Gathering
        fields = [
            "id",
            "title",
            "description",
            "poster",
            "snippet",
            "absolute_url",
            "presenter",
            "address",
            "price",
            "date",
            "max_seats",
            "is_held",
            "is_occupied",
        ]

    # Method to get the absolute URL for a gathering
    def get_abs_url(self, obj):
        request = self.context.get("request")
        return request.build_absolute_uri(obj.pk)

    # Customize the representation of a gathering
    def to_representation(self, instance):
        request = self.context.get("request")
        rep = super().to_representation(instance)
        if instance.price == 0:
            filled_seats = GatheringUser.objects.filter(gathering=instance.id).count()
        else:
            filled_seats = GatheringUser.objects.filter(
                gathering=instance.id, is_paid=True
            ).count()
        if request.parser_context.get("kwargs").get("pk"):
            rep.pop("snippet", None)
            rep.pop("absolute_url", None)
        else:
            rep.pop("description", None)
        rep["filled_seats"] = filled_seats
        rep["empty_seats"] = instance.max_seats - filled_seats
        return rep

# Serializer for editing gathering details
class GatheringEditSerializer(serializers.ModelSerializer):
    class Meta:
        model = Gathering
        fields = [
            "id",
            "title",
            "description",
            "poster",
            "presenter",
            "address",
            "price",
            "date",
            "max_seats",
            "is_held",
            "is_occupied",
        ]

# Serializer for user data (read-only)
class UserDataSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ["id", "phone", "first_name", "last_name", "fullname", "is_ban"]
        read_only_fields = [
            "id",
            "phone",
            "first_name",
            "last_name",
            "fullname",
            "is_ban",
        ]

# Serializer for creating gathering registrations
class GatheringRegistrationCreateSerializer(serializers.ModelSerializer):
    discount_code = serializers.CharField(write_only=True, required=False)
    user = UserDataSerializer(read_only=True)

    class Meta:
        model = GatheringUser
        fields = [
            "id",
            "gathering",
            "user",
            "is_paid",
            "discount_code",
        ]
        read_only_fields = ["id", "is_paid", "user"]

    def to_representation(self, instance):
        rep = super().to_representation(instance)
        rep["detail"] = "successfully added to List"
        return rep

    def validate(self, attrs):
        request = self.context.get("request")

        # Check if the user is already registered for the gathering
        if GatheringUser.objects.filter(
            gathering=attrs.get("gathering"), user=request.user
        ).exists():
            raise serializers.ValidationError(Errors.ALREADY_REGISTERED)

        # Calculate the number of filled seats
        if attrs.get("gathering").price == 0:
            filled_seats = GatheringUser.objects.filter(
                gathering=attrs.get("gathering").id
            ).count()
        else:
            filled_seats = GatheringUser.objects.filter(
                gathering=attrs.get("gathering").id, is_paid=True
            ).count()

        # Validate the discount code
        if attrs.get("discount_code", None):
            try:
                discount_obj = Discount.objects.get(
                    code=attrs.get("discount_code"), gathering=attrs.get("gathering")
                )
            except Discount.DoesNotExist:
                attrs.pop("discount_code", None)
                raise serializers.ValidationError(Errors.INVALID_DISCOUNT_CODE)
            else:
                attrs.pop("discount_code", None)
                attrs["discount"] = discount_obj

        # Validate event details for registration
        if (
            attrs.get("gathering").is_held == False
            and attrs.get("gathering").is_occupied == False
            and int(attrs.get("gathering").max_seats) > filled_seats
            and request.user.is_ban == False
        ):
            pass
        elif attrs.get("gathering").is_held == True:
            raise serializers.ValidationError(Errors.GATHERING_HELD)
        elif attrs.get("gathering").is_occupied == True:
            raise serializers.ValidationError(Errors.Full_capacity)
        elif int(attrs.get("gathering").max_seats) <= filled_seats:
            raise serializers.ValidationError(Errors.Full_capacity)
        elif request.user.is_ban == True:
            raise serializers.ValidationError(Errors.USER_BANNED)
        return super().validate(attrs)

    def create(self, validated_data):
        validated_data["user"] = self.context.get("request").user
        return super().create(validated_data)

# Serializer for canceling gathering registrations
class GatheringRegistrationDeleteSerializer(serializers.ModelSerializer):
    class Meta:
        model = GatheringUser
        fields = [
            "id",
            "gathering",
            "user",
        ]
        read_only_fields = ["id", "gathering", "user"]

    def validate(self, attrs):
        if self.instance.gathering.is_held == True:
            raise serializers.ValidationError(Errors.GATHERING_HELD)
        elif (
            timezone.localtime() + datetime.timedelta(days=2)
            > self.instance.gathering.date
        ):
            raise serializers.ValidationError(Errors.TIME_CANCELLATION)
        return super().validate(attrs)

    def delete(self):
        obj = self.instance
        obj.delete()
        return {"detail": "successfully cancelled"}, status.HTTP_204_NO_CONTENT

# Serializer for gathering registrations (read-only)
class GatheringRegistrationSerializer(serializers.ModelSerializer):
    user = UserDataSerializer(read_only=True)

    class Meta:
        model = GatheringUser
        fields = [
            "id",
            "gathering",
            "user",
            "is_paid",
        ]
        read_only_fields = ["id", "is_paid"]

    def to_representation(self, instance):
        if instance.gathering.price == 0:
            filled_seats = GatheringUser.objects.filter(
                gathering=instance.gathering.id
            ).count()
        else:
            filled_seats = GatheringUser.objects.filter(
                gathering=instance.gathering.id, is_paid=True
            ).count()
        rep = super().to_representation(instance)
        if instance.gathering.price == 0 or instance.is_paid:
            rep["gathering"] = {
                "id": instance.gathering.id,
                "title": instance.gathering.title,
                "price": instance.gathering.price,
                "max_seats": instance.gathering.max_seats,
                "filled_seats": filled_seats,
                "link": instance.gathering.link,
                "uuid": instance.uuid,
                "uuid_url": reverse(
                    "gathering:api-v1:check_in", kwargs={"get_uuid": instance.uuid}
                ),
                "check_in": instance.check_in,
            }
        elif instance.gathering.price != 0 and instance.is_paid == False:
            rep["gathering"] = {
                "id": instance.gathering.id,
                "title": instance.gathering.title,
                "price": instance.gathering.price,
                "max_seats": instance.gathering.max_seats,
                "filled_seats": filled_seats,
            }
            rep["payment_gateway"] = "Activate"
        return rep

# Serializer for gathering photos (read-only)
class GatheringImageSerializer(serializers.ModelSerializer):
    class Meta:
        model = Photo
        fields = ["gathering", "image"]

# Serializer for checking in users at gatherings
class GatheringCheckInSerializer(serializers.ModelSerializer):
    class Meta:
        model = GatheringUser
        fields = ["uuid", "check_in"]
        read_only_fields = ["uuid", "check_in"]

    def to_representation(self, instance):
        rep = super().to_representation(instance)
        rep["accept"] = True
        return rep

    def validate(self, attrs):
        if not self.instance.gathering.is_held and not self.instance.check_in:
            pass
        elif self.instance.gathering.is_held:
            raise serializers.ValidationError(Errors.GATHERING_HELD)
        elif self.instance.check_in:
            raise serializers.ValidationError(Errors.ALREADY_ENTERED)
        return super().validate(attrs)

    def update(self, instance, validated_data):
        instance.check_in = True
        return super().update(instance, validated_data)

# Serializer for gathering discount details (read-only)
class DiscountSerializer(serializers.ModelSerializer):
    gathering = serializers.SerializerMethodField()

    class Meta:
        model = Discount
        fields = ["code", "gathering", "discount_percentage", "status"]

    def get_gathering(self, instance):
        return {
            "id": instance.gathering.id,
            "title": instance.gathering.title,
            "price": instance.gathering.price,
        }

# Serializer for processing payments through a gateway
class GatheringGoGatewaySerializer(serializers.ModelSerializer):
    class Meta:
        model = GatheringUser
        fields = [
            "id",
        ]
        read_only_fields = [
            "id",
        ]

    def validate(self, attrs):
        request = self.context.get("request")
        if self.instance.gathering.price != 0:
            filled_seats = GatheringUser.objects.filter(
                gathering=self.instance.gathering.id
            ).count()
        else:
            raise serializers.ValidationError(Errors.EVENT_IS_FREE)

        if (
            int(self.instance.gathering.max_seats) > filled_seats
            and not self.instance.gathering.is_held
            and not self.instance.gathering.is_occupied
            and not self.instance.is_paid
            and not self.instance.user.is_ban
        ):
            pass
        elif (
            int(self.instance.gathering.max_seats) <= filled_seats
            or self.instance.gathering.is_occupied
        ):
            raise serializers.ValidationError(Errors.Full_capacity)
        elif self.instance.gathering.is_held:
            raise serializers.ValidationError(Errors.GATHERING_HELD)
        elif self.instance.is_paid:
            raise serializers.ValidationError(Errors.ALREADY_IS_PAID_TRUE)
        elif self.instance.user.is_ban:
            raise serializers.ValidationError(Errors.USER_BANNED)

        if self.instance.discount != None:
            attrs["final_price"] = self.instance.gathering.price - (
                (
                    self.instance.gathering.price
                    * self.instance.discount.discount_percentage
                )
                / 100
            )
        else:
            attrs["final_price"] = self.instance.gathering.price

        self.final_price = attrs["final_price"]
        return super().validate(attrs)

    def bank_initialization(self):
        if self.final_price <= 0:
            obj = GatheringUser.objects.get(pk=self.context["pk"])
            obj.is_paid = True
            obj.save()
            return {
                "detail": "The user was successfully registered in the event"
            }, status.HTTP_201_CREATED
        else:
            data = Gateway(request=self.context["request"]).go_to_gateway_view(
                amount=self.final_price, object_id=self.context["pk"]
            )
            if data["success"] == True:
                return data["context"], status.HTTP_200_OK
            elif data["success"] == False and data["object_DoesNotExist"] == False:
                return {"detail": "Error"}, status.HTTP_502_BAD_GATEWAY
            elif data["success"] == False and data["object_DoesNotExist"]:
                return {"detail": "Not found."}, status.HTTP_404_NOT_FOUND

core/gathering/api/v1/serializers.py
H-VICTOOR/TechCafe-Backend
from django.urls import path, include
from gathering.api.v1 import views
from rest_framework.routers import DefaultRouter

app_name = "api-v1"
router = DefaultRouter()

urlpatterns = [
    path(
        "event_register/<int:pk>/",
        views.GatheringRegistrationModelViewSet.as_view(
            {"get": "retrieve", "delete": "destroy"}
        ),
        name="event-register",
    ),
    path(
        "event_register/",
        views.GatheringRegistrationModelViewSet.as_view(
            {"get": "list", "post": "create"}
        ),
        name="my-event-list",
    ),
    path(
        "check_in/<str:get_uuid>/",
        views.GatheringCheckInModelViewSet.as_view({"get": "retrieve"}),
        name="check_in",
    ),
    path(
        "discount/check/<str:get_discount>/",
        views.DiscountModelViewSet.as_view({"get": "retrieve"}),
        name="discount-check",
    ),
    path(
        "event_register/go_to_gateway/<int:pk>/",
        views.GatheringGoGatewayApiView.as_view(),
        name="event-register-go-to-gateway",
    ),
    path(
        "event_register/gateway_callback/",
        views.GatheringGatewayCallbackApiView.as_view(),
        name="event-register-gateway-callback",
    ),
]


router.register(
    "manage-event", views.GatheringEditModelViewSet, basename="manage-event"
)
router.register("event", views.GatheringModelViewSet, basename="event")
router.register("images", views.GatheringImageModelViewSet, basename="images")
urlpatterns += router.urls

core/gathering/api/v1/urls.py
H-VICTOOR/TechCafe-Backend
from gathering.api.v1.serializers import (
    GatheringSerializer,
    GatheringEditSerializer,
    GatheringRegistrationSerializer,
    GatheringImageSerializer,
    DiscountSerializer,
    GatheringGoGatewaySerializer,
    GatheringCheckInSerializer,
    GatheringRegistrationCreateSerializer,
    GatheringRegistrationDeleteSerializer,
)
from gathering.models import Gathering, GatheringUser, Photo, Discount
from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated, IsAdminUser, AllowAny
from rest_framework.response import Response
from rest_framework import status
from rest_framework import filters
from django.db import IntegrityError
from django.shortcuts import get_object_or_404
from django.utils import timezone
import datetime
from django.core.exceptions import ValidationError
from utils.constants import Errors
from rest_framework import generics
from rest_framework.views import APIView
from gathering.services.bank_gateway import Gateway

# ViewSet for managing gathering CRUD operations (staff only)
class GatheringEditModelViewSet(viewsets.ModelViewSet):
    """CRUD options for managing gatherings (staff only)"""

    permission_classes = [IsAdminUser]
    serializer_class = GatheringEditSerializer
    queryset = Gathering.objects.all()

# ViewSet for retrieving gathering details and filtering/searching
class GatheringModelViewSet(viewsets.ReadOnlyModelViewSet):
    """All users can view event details, filter by date, and search for events."""

    permission_classes = [AllowAny]
    serializer_class = GatheringSerializer
    queryset = Gathering.objects.all()
    lookup_field = "pk"
    filter_backends = [filters.SearchFilter, filters.OrderingFilter]
    search_fields = ["title"]
    ordering_fields = ["date", "is_held", "is_occupied"]

# ViewSet for validating discount codes
class DiscountModelViewSet(viewsets.ReadOnlyModelViewSet):
    """Validation for discount codes"""

    permission_classes = [AllowAny]
    serializer_class = DiscountSerializer
    queryset = Discount.objects.filter(status=True)
    lookup_field = "code"
    lookup_url_kwarg = "get_discount"

# ViewSet for gathering registrations (user registration for events)
class GatheringRegistrationModelViewSet(viewsets.ModelViewSet):
    """
    Users can register for events in this class, but they must be logged in to register.
    """

    permission_classes = [IsAuthenticated]

    def get_queryset(self, pk=None):
        """
        The get_queryset function allows us to filter the data received from the database based on our needs.
        """
        if self.action == "list":
            queryset = GatheringUser.objects.filter(user=pk)
        elif self.action == "destroy":
            queryset = get_object_or_404(GatheringUser, id=pk)
        elif self.action == "retrieve":
            queryset = GatheringUser.objects.filter(user=self.request.user)
        return queryset

    def get_serializer_class(self, *args, **kwargs):
        if self.action == "create":
            return GatheringRegistrationCreateSerializer
        elif self.action == "list" or self.action == "retrieve":
            return GatheringRegistrationSerializer
        elif self.action == "destroy":
            return GatheringRegistrationDeleteSerializer

    def list(self, request):
        """Get the list of events registered by the user."""
        serializer = self.serializer_class(
            self.get_queryset(pk=request.user), many=True
        )
        return Response(serializer.data)

    def create(self, request):
        """
        If the method is POST, we create an object for the user in a custom table.
        If the serializer is valid, we return "successfully added to list."
        """
        serializer = self.get_serializer(
            data=request.data, many=False, context={"request": self.request}
        )
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return Response(serializer.data, status=status.HTTP_201_CREATED)

    def destroy(self, request, pk):
        serializer = self.get_serializer(
            data=request.data,
            instance=self.get_queryset(pk=pk),
            context={"request": self.request, "pk": pk},
        )
        serializer.is_valid(raise_exception=True)
        detail, status = serializer.delete()
        return Response(detail, status)

# ViewSet for checking in users at events (staff only)
class GatheringCheckInModelViewSet(viewsets.ReadOnlyModelViewSet):
    permission_classes = [IsAdminUser]
    serializer_class = GatheringCheckInSerializer

    def get_queryset(self, get_uuid=None):
        if self.action == "retrieve":
            try:
                queryset = get_object_or_404(GatheringUser, uuid=get_uuid)
            except ValidationError:
                queryset = False
        return queryset

    def retrieve(self, request, get_uuid, *args, **kwargs):
        query = self.get_queryset(get_uuid=get_uuid)
        if query:
            serializer = self.serializer_class(
                data=request.data,
                instance=query,
            )
            serializer.is_valid(raise_exception=True)
            serializer.save()
            return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
        else:
            return Response(Errors.INVALID_UUID, status=status.HTTP_406_NOT_ACCEPTABLE)

# ViewSet for gathering photos (staff only)
class GatheringImageModelViewSet(viewsets.ReadOnlyModelViewSet):
    """Gathering Photos (staff only)"""

    permission_classes = [AllowAny]
    serializer_class = GatheringImageSerializer
    queryset = Photo.objects.all()
    lookup_field = "pk"
    filter_backends = [filters.SearchFilter]
    search_fields = ["gathering__id"]

# API view for processing payments through a gateway
class GatheringGoGatewayApiView(generics.CreateAPIView):  # UpdateAPIView
    permission_classes = [IsAuthenticated]
    serializer_class = GatheringGoGatewaySerializer
    allowed_methods = ["POST", "OPTIONS"]

    def get_queryset(self, id=None):
        obj = get_object_or_404(GatheringUser, pk=id)
        return obj

    def post(self, request, pk, *args, **kwargs):
        serializer = self.serializer_class(
            data=request.data,
            instance=self.get_queryset(id=pk),
            context={"request": self.request, "pk": pk},
        )
        serializer.is_valid(raise_exception=True)
        result, status = serializer.bank_initialization()
        return Response(result, status)

# API view for handling gateway callback
class GatheringGatewayCallbackApiView(APIView):
    def get(self, request):
        data = Gateway(request=request).callback_gateway_view()
        if data["success"] == True:
            return Response(data["context"], status=status.HTTP_200_OK)
        elif data["success"] == False and data["object_DoesNotExist"]:
            Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND)
        elif data["success"] == False and data["object_DoesNotExist"] == False:
            return Response(
                {"detail": f"{data['error']}"}, status=status.HTTP_502_BAD_GATEWAY
            )

core/gathering/api/v1/views.py
H-VICTOOR/TechCafe-Backend
from django.apps import AppConfig


class GatheringConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "gathering"

core/gathering/apps.py
H-VICTOOR/TechCafe-Backend
from django.db import models
from django.dispatch import receiver
from django.db.models.signals import post_save, pre_save
import os
import uuid
from django.core.exceptions import ValidationError
from django.core.validators import MinValueValidator, MaxValueValidator

# Create your models here.

# Define a model for gatherings.
class Gathering(models.Model):
    # Fields for gathering information.
    title = models.CharField(max_length=250, blank=False, null=False)
    description = models.TextField(blank=False, null=False)
    poster = models.ImageField(upload_to="events/posters")
    presenter = models.ManyToManyField("accounts.User", blank=True)
    address = models.TextField(null=True, blank=True)
    price = models.IntegerField(default=0, blank=False, null=False)
    link = models.URLField(null=True, blank=True)
    date = models.DateTimeField(auto_now=False)
    max_seats = models.PositiveSmallIntegerField(default=0, blank=False, null=False)
    is_online = models.BooleanField(default=False)
    is_held = models.BooleanField(default=False)
    is_occupied = models.BooleanField(default=False)
    created_date = models.DateTimeField(auto_now_add=True)
    updated_date = models.DateTimeField(auto_now=True)

    def __str__(self):
        return self.title

    def get_snippet(self):
        return self.description[0:60] + "..."

    # Custom validation to ensure link field consistency.
    def clean(self):
        if self.is_online and not self.link:
            raise ValidationError(
                "When is_online is set to True, the link field cannot be blank."
            )
        elif not self.is_online and self.link:
            raise ValidationError(
                "When is_online is set to False, the link field cannot have a link."
            )

# Define a model for discounts associated with gatherings.
class Discount(models.Model):
    code = models.CharField(max_length=250, blank=False, null=False, unique=True)
    gathering = models.ForeignKey(
        Gathering, blank=False, null=False, on_delete=models.CASCADE
    )
    discount_percentage = models.PositiveSmallIntegerField(
        default=0,
        validators=[MinValueValidator(0), MaxValueValidator(100)],
        blank=False,
        null=False,
    )
    status = models.BooleanField(default=False)
    created_date = models.DateTimeField(auto_now_add=True)
    updated_date = models.DateTimeField(auto_now=True)

    def __str__(self):
        return f"{self.code}"

# Define a model to associate users with gatherings and discounts.
class GatheringUser(models.Model):
    user = models.ForeignKey(to="accounts.User", on_delete=models.CASCADE)
    gathering = models.ForeignKey(Gathering, on_delete=models.CASCADE)
    discount = models.ForeignKey(
        Discount, blank=True, null=True, on_delete=models.SET_NULL
    )
    bank_gateway = models.ForeignKey(
        "azbankgateways.Bank", blank=True, null=True, on_delete=models.SET_NULL
    )
    uuid = models.UUIDField(unique=True, default=uuid.uuid4, editable=False)
    is_paid = models.BooleanField(default=False)
    check_in = models.BooleanField(default=False)
    created_date = models.DateTimeField(auto_now_add=True)
    updated_date = models.DateTimeField(auto_now=True)

    class Meta:
        unique_together = [
            "gathering",
            "user",
        ]

    def __str__(self):
        return f"{self.user}"

# Define a model for photos associated with gatherings.
class Photo(models.Model):
    gathering = models.ForeignKey(
        Gathering, related_name="gathering_img", on_delete=models.CASCADE
    )
    image = models.ImageField(upload_to="events/gathering")

    def __str__(self):
        return f"{self.gathering.id}"

# Signal handler to delete associated image files when a Photo object is deleted.
@receiver(models.signals.post_delete, sender=Photo)
def auto_delete_file_on_delete(sender, instance, **kwargs):
    if instance.image:
        if os.path.isfile(instance.image.path):
            os.remove(instance.image.path)

# Signal handler to delete unpaid users when a Gathering is held and has a non-zero price.
@receiver(post_save, sender=Gathering)
def delete_unpaid_users(sender, instance, created, **kwargs):
    if not created:
        if instance.is_held and instance.price != 0:
            try:
                GatheringUser.objects.filter(
                    gathering=instance.id, is_paid=False
                ).delete()
            except GatheringUser.DoesNotExist:
                pass

# Signal handler to deactivate discounts when a Gathering is held.
@receiver(post_save, sender=Gathering)
def deactivate_discounts_on_event_held(sender, instance, created, **kwargs):
    if not created:
        if instance.is_held:
            Discount.objects.filter(gathering=instance.id, status=True).update(
                status=False
            )

core/gathering/models.py
H-VICTOOR/TechCafe-Backend
import logging
from django.urls import reverse
from django.shortcuts import render
from azbankgateways import (
    bankfactories,
    models as bank_models,
    default_settings as settings,
)
from azbankgateways.exceptions import AZBankGatewaysException
from gathering.models import GatheringUser
from rest_framework.response import Response
from rest_framework import status


class Gateway:
    def __init__(self, request):
        self.request = request

    def go_to_gateway_view(self, amount, object_id):
        try:
            gathering_user = GatheringUser.objects.get(id=object_id)
        except GatheringUser.DoesNotExist:
            return {
                "success": False,
                "context": None,
                "error": "GatheringUser object Not Found",
                "object_DoesNotExist": True,
            }
        else:
            amount = amount
            user_mobile_number = self.request.user.phone

            factory = bankfactories.BankFactory()
            try:
                bank = (
                    factory.auto_create()
                )  # or factory.create(bank_models.BankType.BMI) or set identifier
                bank.set_request(self.request)
                bank.set_amount(amount)
                # callback url
                bank.set_client_callback_url(
                    reverse("gathering:api-v1:event-register-gateway-callback")
                )  # reverse('callback-gateway')
                bank.set_mobile_number(user_mobile_number)

                # bank record
                bank_record = bank.ready()

                # add bank record for object in GatheringUser
                gathering_user.bank_gateway = bank_record
                gathering_user.save()

                # هدایت کاربر به درگاه بانک
                context = bank.get_gateway()
                return {
                    "success": True,
                    "context": context,
                    "error": None,
                    "object_DoesNotExist": False,
                }
            except AZBankGatewaysException as e:
                logging.critical(e)
                return {
                    "success": False,
                    "context": None,
                    "error": e,
                    "object_DoesNotExist": False,
                }

    def callback_gateway_view(self):
        tracking_code = self.request.GET.get(settings.TRACKING_CODE_QUERY_PARAM, None)
        if not tracking_code:
            return {
                "success": False,
                "context": None,
                "error": "invalid Link",
                "object_DoesNotExist": False,
            }

        try:
            bank_record = bank_models.Bank.objects.get(tracking_code=tracking_code)
        except bank_models.Bank.DoesNotExist:
            logging.debug("invalid Link")
            return {
                "success": False,
                "context": None,
                "error": "invalid Link",
                "object_DoesNotExist": False,
            }

        if bank_record.is_success:
            try:
                gathering_user_obj = GatheringUser.objects.get(bank_gateway=bank_record)
            except GatheringUser.DoesNotExist:
                return {
                    "success": False,
                    "context": None,
                    "error": "Not Found",
                    "object_DoesNotExist": True,
                }
            else:
                gathering_user_obj.is_paid = True
                gathering_user_obj.save()

            return {
                "success": True,
                "context": "Payment was successful",
                "error": None,
                "object_DoesNotExist": False,
            }

        # پرداخت موفق نبوده است. اگر پول کم شده است ظرف مدت ۴۸ ساعت پول به حساب شما بازخواهد گشت.
        return {
            "success": False,
            "context": None,
            "error": "پرداخت با شکست مواجه شده است. اگر پول کم شده است ظرف مدت ۴۸ ساعت پول به حساب شما بازخواهد گشت.",
            "object_DoesNotExist": False,
        }

core/gathering/services/bank_gateway.py
H-VICTOOR/TechCafe-Backend
from django.contrib.auth import get_user_model
from django.test import TestCase
from gathering.models import Gathering
from datetime import datetime


class EventModelsTests(TestCase):
    def setUp(self):
        User = get_user_model()
        self.user_obj = User.objects.create_user(
            phone="00000000000", password="a/@1234567"
        )

    def test_create_event_with_valid_data(self):
        # create Event Object With valid data
        event_obj = Gathering.objects.create(
            title="test_title",
            description="test_description",
            poster="test.png",
            address="test_address",
            price=1.000,
            link="https://www.google.com",
            date=datetime.now(),
            max_seats="5",
            is_online=True,
            is_held=False,
            is_occupied=False,
        )
        # add presenter to event
        event_obj.presenter.add(self.user_obj)

        self.assertEqual(event_obj.presenter.count(), 1)
        self.assertIn(self.user_obj, event_obj.presenter.all())
        self.assertEqual(event_obj.title, "test_title")
        self.assertFalse(event_obj.is_held)
        self.assertFalse(event_obj.is_held)
        self.assertTrue(event_obj.is_online)

core/gathering/tests/test_model.py
H-VICTOOR/TechCafe-Backend
from django.urls import path, include

app_name = "gathering"

urlpatterns = [path("api/v1/", include("gathering.api.v1.urls"))]

core/gathering/urls.py
H-VICTOOR/TechCafe-Backend
class Errors:
    def generate_error(msg):
        return {"detail": msg}

    # ===========================================================  User related errors
    INVALID_PHONE_NUMBER_PATTERN = generate_error(
        "Invalid Phone Number pattern (regex)"
    )
    INVALID_OTP_PATTERN = generate_error("Invalid otp code pattern (regex)")
    INVALID_OTP_CODE = generate_error("Invalid otp code")
    OTP_BLANK = generate_error("OTP cant be blank")
    OTP_SPAM = generate_error("Please wait for a while to resend and try again")
    # ===========================================================  Gathering related errors
    GATHERING_HELD = generate_error("gathering has been held")
    Full_capacity = generate_error("All seats are occupied")
    USER_BANNED = generate_error(
        "User account is limited to register for the event, please contact support"
    )
    ALREADY_REGISTERED = generate_error(
        "The user has already registered for this event"
    )
    TIME_CANCELLATION = generate_error(
        "You cannot cancel your ticket (to cancel the ticket, it must be more than 2 days before the event). If you need to cancel immediately, contact support."
    )
    ALREADY_ENTERED = generate_error("already entered")
    INVALID_UUID = generate_error("Invalid UUID")
    INVALID_DISCOUNT_CODE = generate_error("Invalid discount code")
    ALREADY_IS_PAID_TRUE = generate_error(
        "Payment has already been made successfully for this object"
    )
    EVENT_IS_FREE = generate_error("This event is free and you cannot pay a fee")

    # ===========================================================  system related errors
    XSS_ATTACK_DETECTION = generate_error(
        "XSS protection activate, dont use html tags in fields"
    )
    PASSWORD_MISMATCHED = generate_error("Password mismatched")
    SMS_PANEL = generate_error(
        "There was an error sending a message from the SMS panel"
    )

core/utils/constants.py
H-VICTOOR/TechCafe-Backend
import re
from django.core.exceptions import ValidationError
from django.utils.translation import gettext_lazy as _
from utils.constants import Errors


class FieldValidators:
    def xss_validator(value):
        XSS_PATTERN = r"<(script|img|a|div|iframe|style)[^>]*>"
        if re.search(XSS_PATTERN, value):
            raise ValidationError(
                _(f"{Errors.XSS_ATTACK_DETECTION}"),
            )

    def iranian_phone_validator(value):
        IRANIAN_PHONE_PATTERN = r"^09\d{9}$"
        if not re.match(IRANIAN_PHONE_PATTERN, value):
            raise ValidationError(
                _(f"{Errors.INVALID_PHONE_NUMBER_PATTERN}"),
            )

    def otp_validator(value):
        OTP_PATTERN = r"^[0-9]{4}$"
        if not re.match(OTP_PATTERN, value):
            raise ValidationError(
                _(f"{Errors.INVALID_OTP_PATTERN}"),
            )

core/validators/fieldvalidators.py
H-VICTOOR/TechCafe-Backend
from django.core.exceptions import ValidationError


class VolumeValidator:
    def validate_avatar_size(value):
        limit_mb = 5
        if value.size > limit_mb * 1024 * 1024:  # Convert MB to bytes
            raise ValidationError(f"Avatar size should be {limit_mb} MB or less.")

core/validators/volumevalidator.py
H-VICTOOR/TechCafe-Backend