Django
& DRF

python from rest_framework import serializers from .models import School class SchoolSerializer(serializers.ModelSerializer): school_settings = SettingsSerializer(required=False) class Meta: model = School fields = '__all__' # جميع الحقول exclude = ['vendors'] # استثناء حقول معينة depth = 1 # عمق العلاقات المتداخلة

python class TeacherSerializer(serializers.ModelSerializer): job_type = serializers.SerializerMethodField() status = serializers.SerializerMethodField() full_name = serializers.SerializerMethodField() class Meta: model = Teacher fields = '__all__' def get_status(self, obj): # إرجاع object مخصص بدل قيمة واحدة status_dic = dict(ALLAvailableStatus) color_index = list(status_dic).index(obj.status) return { 'value': obj.status, 'label': status_dic[obj.status], 'color': ALLAvailableStatusColors[color_index] } def get_full_name(self, obj): return f"{obj.first_name} {obj.last_name}"

python class PaymentSerializer(serializers.ModelSerializer): # IntegerField للـ ForeignKey — يستقبل ID فقط student = serializers.IntegerField(required=True, write_only=True) payment_class = serializers.IntegerField(required=False, write_only=True) # ReadOnlyField للقراءة من علاقة student_name = serializers.ReadOnlyField(source='student.name') def create(self, validated_data): DB = self.context['request'].headers.get('DB') # تحويل الـ ID إلى Object validated_data['student'] = SchoolStudents.objects.using(DB).get( id=validated_data.pop('student')) payment = Payments.objects.using(DB).create(**validated_data) return payment

python class EditTeacherSerializer(serializers.ModelSerializer): # ListField لاستقبال array من IDs teacher_class = serializers.ListField( child=serializers.IntegerField(), write_only=True ) subjects = serializers.ListField( child=serializers.IntegerField(), write_only=True ) grades = serializers.ListField( child=serializers.IntegerField(), write_only=True ) def create(self, validated_data): teacher_class = validated_data.pop('teacher_class', []) subjects = validated_data.pop('subjects', []) grades = validated_data.pop('grades', []) record = Teacher.objects.using(DB).create(**validated_data) # ربط الـ ManyToMany record.teacher_class.set(teacher_class) record.subjects.set(subjects) record.grades.set(grades) return record

python from rest_framework.validators import UniqueValidator class SchoolChildSerializer(serializers.ModelSerializer): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) request = self.context.get('request') if request: db = get_db_key(request) # تحديث الـ validators ديناميكيًا self.fields['email'].validators = [ UniqueValidator( queryset=SchoolStudents.objects.using(db).all(), message="Email already exists" ) ]

python class DetailsChildSerializer(serializers.ModelSerializer): # قراءة من علاقة ForeignKey parent_name = serializers.ReadOnlyField(source='parent.name', allow_null=True) parent_email = serializers.ReadOnlyField(source='parent.email', allow_null=True) # Nested Serializer للعلاقة school_details = SchoolSerializer(source='school', read_only=True) # SerializerMethodField للمنطق المخصص school_name = serializers.SerializerMethodField() def get_school_name(self, obj): if obj.school: return obj.school.school_name return None

python from rest_framework import viewsets from rest_framework.authentication import TokenAuthentication from rest_framework.permissions import IsAuthenticated class SchoolViewSet(viewsets.ModelViewSet): authentication_classes = [TokenAuthentication] permission_classes = [IsAuthenticated] queryset = School.objects.all() serializer_class = SchoolSerializer def get_queryset(self): # Multi-database support db = get_db_key(self.request) return School.objects.all().using(db) def create(self, request): serializer = self.get_serializer(data=request.data) if not serializer.is_valid(): return api_response( status=ApiStatus.fail, error=error_serialization(serializer.errors) ) self.perform_create(serializer) return api_response(data=serializer.data) def list(self, request): queryset = self.filter_queryset(self.get_queryset()) # Pagination page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True) return api_response(data=serializer.data)

python from rest_framework.views import APIView from rest_framework.authtoken.models import Token class UserRegister(APIView): def post(self, request): serializer = CreateAccountSerializer(data=request.data) if serializer.is_valid(): serializer.save() account = Account.objects.get(pk=serializer.data['id']) # إنشاء Token للمستخدم token, created = Token.objects.get_or_create(user_id=account.pk) return api_response(data={ **serializer.data, 'token': token.key }) return api_response( error=error_serialization(serializer.errors), status=ApiStatus.fail )

python import django_filters from django.db.models import Q class PaymentsViewSet(viewsets.ModelViewSet): queryset = Payments.objects.all() serializer_class = PaymentSerializer filter_backends = [django_filters.rest_framework.DjangoFilterBackend] def list(self, request): queryset = self.filter_queryset(self.get_queryset()) # Text Search if 'search_text' in request.query_params: queryset = queryset.filter( Q(title__icontains=request.query_params.get('search_text')) | Q(student__name__icontains=request.query_params.get('search_text')) ) # Field Filtering if 'type' in request.query_params: queryset = queryset.filter(type=request.query_params.get('type')) if 'status' in request.query_params: queryset = queryset.filter(status=request.query_params.get('status')) # Date Range if 'date_from' in request.query_params: queryset = queryset.filter(created_at__gte=request.query_params.get('date_from')) queryset = queryset.order_by('-created_at') return api_response(data=PaymentSerializer(queryset, many=True).data)

python from rest_framework.routers import DefaultRouter from django.urls import path, include router = DefaultRouter() router.register(r'schools', SchoolViewSet) router.register(r'teachers', TeacherViewSet) router.register(r'students', StudentViewSet) urlpatterns = [ # APIView routes path('login', UserLogin.as_view()), path('register', UserRegister.as_view()), # ViewSet routes (auto-generated) path('', include(router.urls)), ] # Generated URLs: # GET /schools/ → list() # POST /schools/ → create() # GET /schools/{id}/ → retrieve() # PUT /schools/{id}/ → update() # PATCH /schools/{id}/ → partial_update() # DELETE /schools/{id}/ → destroy()

python from django.contrib.auth.models import AbstractUser, BaseUserManager from django.core.validators import RegexValidator # Custom Manager class CustomUserManager(BaseUserManager): def create_user(self, email, password, **extra_fields): email = self.normalize_email(email) user = self.model(email=email, **extra_fields) user.set_password(password) user.save() return user # Validators phone_regex = RegexValidator( regex=r'^\+?1?\d{9,15}$', message="Phone number format: '+999999999'. Up to 15 digits." ) password_regex = RegexValidator( regex=r'^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[@$.!%*?&])[A-Za-z\d@$.!%*?&]{8,16}$', message="8-16 chars, uppercase, lowercase, number, special char" ) # Custom User Model class Account(AbstractUser): username = None # إزالة username email = models.EmailField(unique=True) phone = models.CharField(validators=[phone_regex], max_length=17, unique=True) password = models.CharField(max_length=300, validators=[password_regex], null=True) # Social Login IDs google_id = models.CharField(max_length=500, unique=True, null=True) apple_id = models.CharField(max_length=500, unique=True, null=True) USERNAME_FIELD = 'email' REQUIRED_FIELDS = [] objects = CustomUserManager() def get_full_name(self): if self.first_name and self.last_name: return f"{self.first_name} {self.last_name}" return self.name or ""

python from django.db.models import UniqueConstraint, Q class SchoolClass(models.Model): title = models.CharField(max_length=100) class_grade = models.ForeignKey(Grade, on_delete=models.CASCADE) key = models.CharField(max_length=50, unique=True) class Meta: db_table = "class" # Custom table name ordering = ['-created_at'] # Default ordering # Unique Constraints constraints = [ UniqueConstraint( fields=['title', 'class_grade'], name='class_unique_error' ), # Conditional Unique UniqueConstraint( fields=['email'], condition=Q(is_event_student=False), name='unique_email_if_not_event' ) ] # Indexes indexes = [ models.Index(fields=['key']), models.Index(fields=['created_at']), ]

python # Custom Manager — يفلتر البيانات تلقائيًا class ActivePaymentManager(models.Manager): def get_queryset(self): return super().get_queryset().filter(removed=False) class AllPaymentManager(models.Manager): def get_queryset(self): return super().get_queryset() # جميع البيانات # Base Model class Payments(models.Model): title = models.CharField(max_length=200) amount = models.DecimalField(max_digits=10, decimal_places=2) status = models.CharField(max_length=50) removed = models.BooleanField(default=False) objects = ActivePaymentManager() # Default: excludes removed # Proxy Model — نفس الـ table، manager مختلف class PaymentAllStatus(Payments): class Meta: proxy = True objects = AllPaymentManager() # All records including removed # Usage: # Payments.objects.all() → only active payments # PaymentAllStatus.objects.all() → all payments including removed

python from django.db.models.signals import post_save from django.dispatch import receiver from rest_framework.authtoken.models import Token @receiver(post_save, sender=settings.AUTH_USER_MODEL) def create_auth_token(sender, instance=None, created=False, **kwargs): """إنشاء Token تلقائيًا عند إنشاء مستخدم جديد""" if created: try: Token.objects.get_or_create(user=instance) except Exception as e: logger.exception("Failed to create token: %s", e)

python # Dynamic path based on instance def user_directory_path(instance, filename): return f'user_{instance.id}/{filename}' def teacher_directory_path(instance, filename): return f'teacher_{instance.id}/{filename}' class User(models.Model): user_image = models.ImageField( upload_to=user_directory_path, null=True, blank=True ) class Task(models.Model): def task_file_path(self, filename): return f'tasks_file_{self.pk}/{filename}' file = models.FileField(upload_to=task_file_path, null=True)

python # settings.py — Multiple Databases DATABASES = { 'default': { 'ENGINE': 'django.db.backends.sqlite3', 'NAME': BASE_DIR / 'db.sqlite3', }, 'school_db_1': { 'ENGINE': 'django.db.backends.mysql', 'NAME': 'school1', 'USER': 'root', 'PASSWORD': 'password', 'HOST': 'localhost', }, 'school_db_2': { 'ENGINE': 'django.db.backends.mysql', 'NAME': 'school2', 'USER': 'root', } } DATABASE_ROUTERS = ['cogens.routers.DatabaseRouter']

python # routers.py — Database Router class DatabaseRouter: EXCLUDE_MODELS = ["account", "school", "token"] ONLY_DEFAULT = ["oauth_app", "config_system_setting"] def allow_migrate(self, db, app_label, model_name=None, **hints): if app_label in self.ONLY_DEFAULT: return db == 'default' if db == 'default' and model_name not in self.EXCLUDE_MODELS: return False return True # Usage in Views/Serializers: db = request.headers.get('DB', 'default') queryset = Model.objects.using(db).all()

python # Basic Queries Product.objects.all() # جميع السجلات Product.objects.filter(status='active') # فلترة Product.objects.exclude(removed=True) # استثناء Product.objects.get(id=1) # سجل واحد # Chaining Product.objects.filter(status='active').exclude(price=0).order_by('-created_at') # Q Objects (OR conditions) from django.db.models import Q Product.objects.filter( Q(title__icontains='phone') | Q(description__icontains='phone') ) # Aggregation from django.db.models import Count, Sum, Avg Order.objects.aggregate( total=Sum('amount'), avg_amount=Avg('amount'), count=Count('id') ) # Select Related (ForeignKey optimization) Payment.objects.select_related('student', 'student__school') # Prefetch Related (ManyToMany optimization) Teacher.objects.prefetch_related('subjects', 'teacher_class') # Values & Values List User.objects.values('id', 'email') # → [{'id': 1, 'email': '...'}] User.objects.values_list('id', flat=True) # → [1, 2, 3, ...]

python from django.db import transaction # Atomic Transaction def generate_unique_reference(db_key): for _ in range(10): code = ''.join(random.choices(string.ascii_uppercase, k=6)) try: with transaction.atomic(using=db_key): if not Payments.objects.using(db_key).filter(reference=code).exists(): return code except Exception: continue return uuid.uuid4().hex[:6].upper()

python # settings.py REST_FRAMEWORK = { 'DEFAULT_AUTHENTICATION_CLASSES': ( 'rest_framework.authentication.TokenAuthentication', ), 'DEFAULT_PERMISSION_CLASSES': [ 'rest_framework.permissions.IsAuthenticated' ] } INSTALLED_APPS = [ ... 'rest_framework.authtoken', ]

python # backends.py — Email & Phone Authentication from django.contrib.auth.hashers import check_password from django.contrib.auth import get_user_model class MyAuthBackend: @staticmethod def authenticate(email=None, phone=None, password=None, **kwargs): try: if email: user = get_user_model().objects.get(email=email) elif phone: user = get_user_model().objects.get(phone=phone) else: return None # Check main password or alternative if check_password(password, user.password): return user elif check_password(password, user.alternative_password): return user except Exception: return None # settings.py AUTHENTICATION_BACKENDS = ( 'main.accounts.backends.MyAuthBackend', 'django.contrib.auth.backends.ModelBackend', )

python from google.oauth2 import id_token from google.auth.transport import requests class UserRegister(APIView): def post(self, request): social_provider = request.data.get('social_provider') if social_provider == 'google': google_token = request.data.get('google_token') try: idinfo = id_token.verify_oauth2_token( google_token, requests.Request() ) # Get user info from Google email = idinfo.get('email') google_id = idinfo.get('sub') name = idinfo.get('name') account, created = Account.objects.get_or_create( google_id=google_id, defaults={'email': email, 'name': name, 'email_verified': True} ) except ValueError: return api_response(error="Invalid Google token", status=ApiStatus.fail)

python from rest_framework.permissions import IsAuthenticated, AllowAny, IsAdminUser class MyViewSet(viewsets.ModelViewSet): # View-level permissions permission_classes = [IsAuthenticated] # Method-specific permissions def get_permissions(self): if self.action == 'list': return [AllowAny()] elif self.action == 'destroy': return [IsAdminUser()] return [IsAuthenticated()] # Custom Permission from rest_framework.permissions import BasePermission class IsOwner(BasePermission): def has_object_permission(self, request, view, obj): return obj.created_by == request.user

python # management/commands/alert_attendance.py from django.core.management.base import BaseCommand class Command(BaseCommand): help = 'Alert staff to take attendance' def handle(self, *args, **options): schools = School.objects.all() for school in schools: # Custom timezone handling timezone = pytz.timezone(school.timezone) now = datetime.now(timezone) current_time = now.strftime("%H:%M") if current_time == school.attendance_time: send_notification( user_id=account.pk, db_key=school.db_key, title=f"Attendance Check", body=school.school_name, notification_type=NotificationType.alertAttendance.value ) # Run command: # python manage.py alert_attendance

python # settings.py — Cron Jobs Configuration CRONJOBS = [ ('*/5 * * * *', 'django.core.management.call_command', ['interc_payment']), ('*/5 * * * *', 'django.core.management.call_command', ['alert_attendance']), ('*/30 * * * *', 'django.core.management.call_command', ['late_pickup']), ('0 9,18 * * *', 'django.core.management.call_command', ['future_combo_meals']), ] # Cron Syntax: minute hour day month weekday # */5 = every 5 minutes # 0 9,18 = at 9:00 and 18:00 # * * * = every day

python import uuid import json class StatusTypes: PENDING = 0 SUCCEED = 1 FAILED = 2 class ScheduleJob(models.Model): public_id = models.UUIDField(default=uuid.uuid4) status = models.SmallIntegerField(default=StatusTypes.PENDING) job = models.SmallIntegerField() unique_key = models.CharField(max_length=255) # Prevents duplicates failed_tries = models.SmallIntegerField(default=0) max_tries = models.SmallIntegerField(default=1) db_key = models.CharField(max_length=50, null=True) payload = models.TextField(null=True) # JSON data exception_error = models.TextField(null=True) class Meta: db_table = "schedule_jobs" unique_together = ('status', 'job', 'unique_key') def get_payload(self): return json.loads(self.payload) if self.payload else {} def set_fail(self, exception): self.exception_error = str(exception) self.failed_tries += 1 if self.failed_tries >= self.max_tries: self.status = StatusTypes.FAILED self.save()

python from django.core.mail.backends.smtp import EmailBackend from django.core.mail import EmailMessage from django.template.loader import render_to_string class SendEmailMessage: def __init__(self, subject, emails, DB=None): # Dynamic email config per school email_config = get_setting(DB, ['EMAIL_CONFIG']) self.email_backend = EmailBackend( host=email_config['EMAIL_HOST'], port=email_config['EMAIL_PORT'], username=email_config['EMAIL_HOST_USER'], password=email_config['EMAIL_HOST_PASSWORD'], use_tls=True ) self.subject = subject self.emails = emails def send_email(self, html_message): message = EmailMessage( self.subject, html_message, self.email_user, self.emails, connection=self.email_backend ) message.content_subtype = 'html' message.send() # Usage with Template def send_invoice_email(email, invoice_data, school_key): html_message = render_to_string('invoice_email.html', { 'body': invoice_data, 'school': school.school_name, }) message = SendEmailMessage(DB=school_key, subject='Invoice', emails=[email]) message.send_email(html_message=html_message)

python import enum from pyfcm import FCMNotification class NotificationType(enum.Enum): Login = 'new_login' newMessage = 'new_message' publicAlert = 'public_alert' billing = 'billing' todo = 'todo' alertAttendance = 'alert_attendance' def send_notification(user_id, title, body, notification_type, data, db_key): # Check notification settings settings = Settings.objects.using(db_key).filter(account=user_id) if settings and not settings[0].send_notification: return # Initialize Firebase fcm = FCMNotification( service_account_file="firebase-adminsdk.json", project_id="my-project" ) # Get all user devices devices = Device.objects.filter(account_id=account.pk) all_tokens = [d.device_token for d in devices] # Send to multiple devices params_list = [{ "fcm_token": token, "notification_title": title, "notification_body": body, "data_payload": {'data': json.dumps(data)} } for token in all_tokens] fcm.async_notify_multiple_devices(params_list=params_list)

python # settings.py MIDDLEWARE = [ 'corsheaders.middleware.CorsMiddleware', 'django.middleware.security.SecurityMiddleware', 'django.contrib.sessions.middleware.SessionMiddleware', 'django.middleware.common.CommonMiddleware', 'django_middleware_global_request.middleware.GlobalRequestMiddleware', ] # CORS Configuration CORS_ALLOW_ALL_ORIGINS = True CORS_ALLOW_CREDENTIALS = True CORS_ALLOW_HEADERS = ['*'] # Global Request Pattern from django_middleware_global_request import get_request def save_model_date(instance, request): if request and request.user.is_authenticated: if instance.pk is None: instance.created_by = request.user else: instance.updated_by = request.user instance.updated_at = datetime.now()

python import enum from rest_framework.response import Response from rest_framework import status as stat class ApiStatus(enum.Enum): success = 200 fail = 400 notFound = 404 unauthorized = 401 applicationException = 500 def api_response(data=None, status=ApiStatus.success, message=None, error=None, count=None): """Standardized API response wrapper""" if error: status = ApiStatus.fail return Response({ "data": data, "message": message, "status": status.name, # 'success', 'fail', etc. "errors": error, "count": count }, status=stat.HTTP_200_OK) # Response Examples: # Success: {"data": {...}, "status": "success", "errors": null} # Error: {"data": null, "status": "fail", "errors": {"email": "Required"}}

python def error_serialization(error): """Convert nested DRF errors to flat dict""" # Input: {"email": ["This field is required"]} # Output: {"email": "This field is required"} # Also handles nested: # Input: {"items": [{"student": ["Invalid pk"]}]} # Output: {"items": "student: Invalid pk"} data = {} for item in error: value = error[item] if isinstance(value, list) and len(value) > 0: if isinstance(value[0], str): data[item] = value[0] elif isinstance(value[0], dict): nested_errors = [] for nested_dict in value: for key, val in nested_dict.items(): if isinstance(val, list): nested_errors.append(f"{key}: {val[0]}") data[item] = "; ".join(nested_errors) return dict(data)

python # cogens/urls.py — Main URL Configuration urlpatterns = [ # Auth APIs path('api/', include('main.accounts.urls')), path('api/', include('main.devices.urls')), # School Management path('api/', include('school.school.urls')), path('api/', include('school.teacher.urls')), path('api/', include('school.school_students.urls')), # Staff Management path('api/', include('staff.task.urls')), # Billing & Payments path('api/', include('shopping_billing_promo_code.payment.urls')), # API Versioning path('api/v5/', include('shopping_billing_promo_code.hotlunch.orders.v_5_staff_order.urls')), ]

python # settings.py REST_FRAMEWORK = { 'DEFAULT_FILTER_BACKENDS': [ 'django_filters.rest_framework.DjangoFilterBackend' ], 'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.LimitOffsetPagination', 'PAGE_SIZE': 20, } # Custom Pagination Class from rest_framework.pagination import PageNumberPagination class CustomPagination(PageNumberPagination): page_size = 20 page_size_query_param = 'page_size' max_page_size = 100

python # constants.py — Centralized Status Constants class BillingStatus: PENDING = 'pending' CASH = 'cash_pending' PAID = 'paid' PARTIAL = 'partial' OVER_PAID = 'over_paid' class PaymentTypes: LATE_PICKUP = 'Late Pickup' TUITION = 'Tuition' DEFAULT = 'Other' # Model Choices GENDER_CHOICES = ( ('male', 'male'), ('female', 'female'), ('prefer_not', 'prefer not to say'), ) LOGIN_AS = ( ('school', 'school'), ('employee', 'employee'), ('parent', 'parent'), ('student', 'student'), ('teacher', 'teacher'), ) # Usage in Model gender = models.CharField(max_length=200, choices=GENDER_CHOICES, null=True)