From c2abcd7b970f4c196b63d0c00539d3663865fdc9 Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Sun, 14 Jun 2026 00:03:57 +0330 Subject: [PATCH] feat(payments): add discount code admin API --- apps/payments/api/schemas.py | 38 ++++++++++ apps/payments/api/views.py | 134 ++++++++++++++++++++++++++++++++++- 2 files changed, 170 insertions(+), 2 deletions(-) diff --git a/apps/payments/api/schemas.py b/apps/payments/api/schemas.py index 7b7b274..a20b26b 100644 --- a/apps/payments/api/schemas.py +++ b/apps/payments/api/schemas.py @@ -1,4 +1,42 @@ from ninja import Schema +from datetime import datetime + + +class DiscountCodeSchema(Schema): + id: int + code: str + type: str + value: int + max_discount: int | None = None + is_active: bool + starts_at: datetime | None = None + ends_at: datetime | None = None + usage_limit_total: int | None = None + usage_limit_per_user: int | None = None + min_amount: int | None = None + applicable_event_ids: list[int] + usage_count: int = 0 + created_at: datetime + updated_at: datetime + + +class PagedDiscountCodeSchema(Schema): + count: int + results: list[DiscountCodeSchema] + + +class DiscountCodeWriteSchema(Schema): + code: str + type: str = "percent" + value: int + max_discount: int | None = None + is_active: bool = True + starts_at: datetime | None = None + ends_at: datetime | None = None + usage_limit_total: int | None = None + usage_limit_per_user: int | None = None + min_amount: int | None = None + applicable_event_ids: list[int] = [] class CreatePaymentIn(Schema): diff --git a/apps/payments/api/views.py b/apps/payments/api/views.py index 1106983..8cc4efa 100644 --- a/apps/payments/api/views.py +++ b/apps/payments/api/views.py @@ -1,8 +1,9 @@ from django.conf import settings from django.shortcuts import redirect, get_object_or_404 from django.utils import timezone +from django.db.models import Count, Q -from ninja import Router +from ninja import Query, Router from ninja.errors import HttpError import requests @@ -11,11 +12,140 @@ from apps.events.models import Event, Registration from apps.notifications.services import notify_user from apps.users.tasks import send_critical_sms from core.authentication import jwt_auth -from apps.payments.api.schemas import CouponVerifyIn, CouponVerifyOut, CreatePaymentIn, CreatePaymentOut, PaymentDetailOut +from core.api.schemas import ErrorSchema, MessageSchema +from apps.payments.api.schemas import ( + CouponVerifyIn, + CouponVerifyOut, + CreatePaymentIn, + CreatePaymentOut, + DiscountCodeSchema, + DiscountCodeWriteSchema, + PagedDiscountCodeSchema, + PaymentDetailOut, +) payments_router = Router(tags=["Payments"]) +def _staff_required(user): + return bool(user and (user.is_staff or user.is_superuser)) + + +def _discount_payload(code: DiscountCode): + return { + "id": code.id, + "code": code.code, + "type": code.type, + "value": code.value, + "max_discount": code.max_discount, + "is_active": code.is_active, + "starts_at": code.starts_at, + "ends_at": code.ends_at, + "usage_limit_total": code.usage_limit_total, + "usage_limit_per_user": code.usage_limit_per_user, + "min_amount": code.min_amount, + "applicable_event_ids": list(code.applicable_events.values_list("id", flat=True)), + "usage_count": getattr(code, "usage_count", None) or code.payments.filter( + status__in=[Payment.OrderStatusChoices.PAID, Payment.OrderStatusChoices.PENDING] + ).count(), + "created_at": code.created_at, + "updated_at": code.updated_at, + } + + +def _apply_discount_payload(instance: DiscountCode, payload: DiscountCodeWriteSchema): + data = payload.dict() + event_ids = data.pop("applicable_event_ids", []) + for field, value in data.items(): + setattr(instance, field, value) + instance.code = instance.code.strip().upper() + instance.full_clean() + instance.save() + instance.applicable_events.set(Event.objects.filter(id__in=event_ids, is_deleted=False)) + return instance + + +@payments_router.get("/admin/discount-codes", response={200: PagedDiscountCodeSchema, 403: ErrorSchema}, auth=jwt_auth) +def admin_list_discount_codes( + request, + search: str | None = Query(None), + is_active: bool | None = Query(None), + type: str | None = Query(None), + limit: int = Query(20, ge=1, le=100), + offset: int = Query(0, ge=0), +): + if not _staff_required(request.auth): + return 403, {"error": "Permission denied"} + queryset = DiscountCode.objects.annotate( + usage_count=Count( + "payments", + filter=Q( + payments__status__in=[ + Payment.OrderStatusChoices.PAID, + Payment.OrderStatusChoices.PENDING, + ] + ), + ) + ).prefetch_related("applicable_events").order_by("-created_at") + if search: + queryset = queryset.filter(code__icontains=search) + if is_active is not None: + queryset = queryset.filter(is_active=is_active) + if type: + queryset = queryset.filter(type=type) + count = queryset.count() + return 200, {"count": count, "results": [_discount_payload(item) for item in queryset[offset : offset + limit]]} + + +@payments_router.post("/admin/discount-codes", response={201: DiscountCodeSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) +def admin_create_discount_code(request, payload: DiscountCodeWriteSchema): + if not _staff_required(request.auth): + return 403, {"error": "Permission denied"} + if DiscountCode.all_objects.filter(code=payload.code.strip().upper()).exists(): + return 400, {"error": "Discount code already exists"} + try: + code = _apply_discount_payload(DiscountCode(), payload) + except Exception as exc: + return 400, {"error": str(exc)} + return 201, _discount_payload(code) + + +@payments_router.put("/admin/discount-codes/{int:code_id}", response={200: DiscountCodeSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) +def admin_update_discount_code(request, code_id: int, payload: DiscountCodeWriteSchema): + if not _staff_required(request.auth): + return 403, {"error": "Permission denied"} + code = get_object_or_404(DiscountCode, id=code_id, is_deleted=False) + normalized = payload.code.strip().upper() + if DiscountCode.all_objects.filter(code=normalized).exclude(id=code_id).exists(): + return 400, {"error": "Discount code already exists"} + try: + code = _apply_discount_payload(code, payload) + except Exception as exc: + return 400, {"error": str(exc)} + return 200, _discount_payload(code) + + +@payments_router.delete("/admin/discount-codes/{int:code_id}", response={200: MessageSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) +def admin_delete_discount_code(request, code_id: int): + if not request.auth.is_superuser: + return 403, {"error": "Only superusers can delete discount codes"} + code = get_object_or_404(DiscountCode, id=code_id, is_deleted=False) + code.delete() + return 200, {"message": "Discount code deleted"} + + +@payments_router.post("/admin/discount-codes/{int:code_id}/restore", response={200: MessageSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) +def admin_restore_discount_code(request, code_id: int): + if not request.auth.is_superuser: + return 403, {"error": "Only superusers can restore discount codes"} + try: + code = DiscountCode.deleted_objects.get(id=code_id) + except DiscountCode.DoesNotExist: + return 400, {"error": "Discount code not found"} + code.restore() + return 200, {"message": "Discount code restored"} + + def _event_action_url(event: Event) -> str: root = getattr(settings, "FRONTEND_ROOT", "/") or "/" if not root.endswith("/"):