basic promotion handler

apps/promotions/admin.py

@@ -0,0 +1,10 @@
+from functools import update_wrapper
+
+from django.contrib import admin
+
+from .models import Promotion, Plan
+
+
+admin.site.register(Plan)
+admin.site.register(Promotion)
+

apps/promotions/apps.py

@@ -1,6 +1,6 @@
 from django.apps import AppConfig
 
 
-class SampleConfig(AppConfig):
+class PromotionsConfig(AppConfig):
     default_auto_field = 'django.db.models.BigAutoField'
-    name = 'apps.sample'
+    name = 'apps.promotions'

apps/promotions/handlers.py

@@ -0,0 +1,43 @@
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+from enum import property as enum_property  # ??
+
+
+class PromotionTypeChoices(models.TextChoices):
+    WALLET_PERCENTAGE = 'wallet percentage', _('Wallet percentage')
+    OTHERS = 'others', _('others')
+
+    @enum_property
+    def handler(self):
+        if self.name == self.WALLET_PERCENTAGE.name:
+            return WalletPercentageHandler
+
+        elif self.name == self.OTHERS.name:
+            return OtherHandler
+
+        else:
+            raise NotImplementedError()
+
+
+class BasePromotionHandler:
+    def __init__(self, plan):
+        self.plan = plan
+
+    def calculate(self, **kwargs):
+        raise NotImplementedError()
+
+
+class WalletPercentageHandler(BasePromotionHandler):
+    def calculate(self, **kwargs):
+        base_amount = kwargs.get('base_amount', 0)
+
+        if base_amount <= 1000 and self.plan.balance >= 1000:
+            promotion_amount = base_amount
+
+        else:
+            promotion_amount = 1000 if self.plan.balance >= 1000 else self.plan.balance
+        return promotion_amount
+
+
+class OtherHandler(BasePromotionHandler):
+    pass

apps/promotions/migrations/0001_initial.py

@@ -0,0 +1,52 @@
+# Generated by Django 5.2.4 on 2025-08-02 16:18
+
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        migrations.swappable_dependency(settings.OAUTH2_PROVIDER_APPLICATION_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Plan',
+            fields=[
+                ('uuid', models.UUIDField(db_index=True, default=uuid.uuid4, editable=False, primary_key=True, serialize=False, unique=True)),
+                ('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
+                ('updated_at', models.DateTimeField(auto_now=True, db_index=True)),
+                ('title', models.CharField(db_index=True, max_length=255, verbose_name='title')),
+                ('description', models.TextField(blank=True, verbose_name='description')),
+                ('promotion_type', models.CharField(choices=[('wallet percentage', 'Wallet percentage'), ('others', 'others')], db_index=True, max_length=64, verbose_name='promotion type')),
+                ('policy', models.JSONField(blank=True, default=dict, null=True, verbose_name='policy')),
+                ('balance', models.DecimalField(decimal_places=2, default=0, max_digits=10, verbose_name='balance')),
+                ('application', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='+', to=settings.OAUTH2_PROVIDER_APPLICATION_MODEL)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='Promotion',
+            fields=[
+                ('uuid', models.UUIDField(db_index=True, default=uuid.uuid4, editable=False, primary_key=True, serialize=False, unique=True)),
+                ('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
+                ('updated_at', models.DateTimeField(auto_now=True, db_index=True)),
+                ('state', models.IntegerField(choices=[(0, 'Created'), (1, 'Assigned'), (10, 'Expired')], default=0)),
+                ('data', models.JSONField(blank=True, null=True)),
+                ('application', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='+', to=settings.OAUTH2_PROVIDER_APPLICATION_MODEL)),
+                ('plan', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='promotions', to='promotions.plan')),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+    ]

apps/promotions/models.py

@@ -0,0 +1,68 @@
+from django.conf import settings
+from django.db import models
+from oauth2_provider.settings import oauth2_settings
+
+from apps.promotions.handlers import PromotionTypeChoices, BasePromotionHandler
+from utils.models import BaseModel
+
+from django.utils.translation import gettext_lazy as _
+
+
+class PlanQuerySet(models.QuerySet):
+    def calculate(self, user, application=None):
+        for plan in self.filter(application=application)[:10]:
+            if not Promotion.objects.used(user):
+                plan.calculate(user)
+
+
+class Plan(BaseModel):
+    # user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
+    application = models.ForeignKey(oauth2_settings.APPLICATION_MODEL, on_delete=models.PROTECT,
+                                    related_name='+', null=True, blank=True)
+    title = models.CharField(max_length=255, verbose_name=_('title'), db_index=True)
+    description = models.TextField(verbose_name=_('description'), blank=True)
+    promotion_type = models.CharField(max_length=64, verbose_name=_('promotion type'), db_index=True, choices=PromotionTypeChoices.choices)
+    policy = models.JSONField(default=dict, verbose_name=_('policy'), null=True, blank=True)
+    balance = models.DecimalField(max_digits=10, decimal_places=2, verbose_name=_('balance'), default=0)
+
+    objects = PlanQuerySet.as_manager()
+
+    def __str__(self):
+        return self.title
+
+    @property
+    def handler(self) -> BasePromotionHandler:
+        return PromotionTypeChoices(self.promotion_type).handler(
+            application=self.application,
+            policy=self.policy,
+            balance=self.balance,
+        )
+
+    def calculate_promotion(self, **kwargs):
+        handler = self.handler
+        self.base_amount = kwargs.get('base_amount', 0)
+        self.promotion_amount = handler.calculate(**kwargs)
+        return self.promotion_amount
+
+
+class PromotionQuerySet(models.QuerySet):
+    def used(self, user):
+        return self.filter(user=user, state__in=[Promotion.StateChoices.CREATED])
+
+
+class Promotion(BaseModel):
+    class StateChoices(models.IntegerChoices):
+        CREATED = 0, _('Created')
+        ASSIGNED = 1, _('Assigned')
+        EXPIRED = 10, _('Expired')
+
+    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
+    application = models.ForeignKey(oauth2_settings.APPLICATION_MODEL, on_delete=models.PROTECT,
+                                    related_name='+', null=True, blank=True)
+    plan = models.ForeignKey(Plan, on_delete=models.PROTECT, related_name='promotions', null=True, blank=True)
+    state = models.IntegerField(choices=StateChoices.choices, default=StateChoices.CREATED)
+
+    data = models.JSONField(null=True, blank=True)
+
+    objects = PromotionQuerySet.as_manager()
+

apps/promotions/serializers.py

@@ -0,0 +1,47 @@
+from rest_framework import serializers
+
+from .models import Promotion, Plan
+
+
+class PlanSerializer(serializers.ModelSerializer):
+    promotion_amount = serializers.SerializerMethodField(method_name='get_promotion_amount')
+
+    def get_promotion_amount(self, obj):
+        user = self.context['request'].user
+        return obj.calculate_promotion(user)
+
+    class Meta:
+        model = Plan
+        fields = (
+            "promotion_amount",
+            "title",
+            "description",
+            "application",
+        )
+        read_only_fields = ('application', "promotion_amount")
+
+class PromotionSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Promotion
+        fields = (
+            "user",
+            "application",
+        )
+        read_only_fields = ('user','application')
+
+class CalculatedPlanSerializer(serializers.ModelSerializer):
+    base_amount = serializers.IntegerField(required=True)
+    promotion_amount = serializers.IntegerField(read_only=True)
+    class Meta:
+        model = Promotion
+        fields = (
+            "base_amount",
+            "user",
+            "application",
+            "promotion_amount"
+        )
+        read_only_fields = ('user','application', "promotion_amount")
+
+    # def get_promotion_amount(self, obj):
+    #     user = self.context['request'].user
+    #     return obj.calculate_promotion(user)

apps/promotions/urls.py

@@ -5,13 +5,13 @@ from utils.router import ProfileRouter
 from . import views
 from django.urls import NoReverseMatch, path, re_path, include
 
-app_name = 'sample'
+app_name = 'promotions'
 
 application_router = DefaultRouter()
-application_router.register(r'', views.ApplicationSampleViewSet, basename='application' )
+application_router.register(r'', views.ApplicationPlanViewSet, basename='application' )
 
-user_router = ProfileRouter()
+user_router = DefaultRouter()
-user_router.register('', views.UserSampleViewSet, basename='user')
+user_router.register('', views.UserPlanViewSet, basename='user')
 
 urlpatterns = [
     path('api/v1/application/<user_uuid>/', include(application_router.urls), name='application'),

apps/promotions/views.py

@@ -0,0 +1,84 @@
+from django.shortcuts import render
+
+from django.http import Http404
+from django.shortcuts import render
+from drf_spectacular.utils import extend_schema
+from oauth2_provider.contrib.rest_framework import TokenMatchesOASRequirements
+from rest_framework import viewsets, mixins
+from rest_framework.decorators import action
+from rest_framework.generics import get_object_or_404
+from rest_framework.viewsets import GenericViewSet
+
+from .models import Promotion, Plan
+from .serializers import PlanSerializer, PromotionSerializer, CalculatedPlanSerializer
+from apps.gooyal_oauth2.models import Application
+from apps.gooyal_oauth2.rest_framework import IsAuthenticatedOrTokenMatchesOASRequirements
+from apps.gooyal_oauth2.utils import get_application
+from ..users.models import User
+
+
+from django.shortcuts import render
+
+from django.http import Http404
+from django.shortcuts import render
+from django.urls import reverse
+from oauth2_provider.contrib.rest_framework import TokenMatchesOASRequirements
+from rest_framework import viewsets, mixins
+from rest_framework.generics import get_object_or_404
+from rest_framework.viewsets import GenericViewSet
+
+from .models import Plan, Promotion
+from apps.gooyal_oauth2.models import Application
+from apps.gooyal_oauth2.rest_framework import IsAuthenticatedOrTokenMatchesOASRequirements
+from apps.gooyal_oauth2.utils import get_application
+from ..users.models import User
+
+
+class ApplicationPlanViewSet(mixins.ListModelMixin,
+                      mixins.CreateModelMixin,
+                      GenericViewSet):
+    queryset = Plan.objects.all()
+    serializer_class = PlanSerializer
+
+    permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements]
+    required_alternate_scopes = {
+        "POST": [["promotions.application.user-plans:submit"]],
+        "GET": [["promotions.application.user-promotions:list-retrieve"]],
+    }
+
+    # def perform_create(self, serializer):
+    #     user_uuid = self.kwargs.get('user_uuid')
+    #     user = User.objects.get(uuid=user_uuid)
+    #     application = get_application(self.request)
+    #     instance = serializer.save(user=user, application=application)
+
+
+class UserPlanViewSet(mixins.RetrieveModelMixin,
+                      mixins.ListModelMixin,
+                          GenericViewSet):
+
+    serializer_class = PlanSerializer
+    permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements]
+    required_alternate_scopes = {
+        "GET": [["promotions.user.self-plans:list-retrieve"]],
+    }
+
+    def get_queryset(self):
+        user = self.request.user
+        application = get_application(self.request)
+
+        queryset = Plan.objects.all()
+        return queryset
+
+    @extend_schema(responses=CalculatedPlanSerializer(many=True))
+    @action(detail=False, methods=['POST'], serializer_class=CalculatedPlanSerializer)
+    def calculate(self, request):
+        user = self.request.user
+        serializer = self.get_serializer(data=self.request.data)
+        serializer.is_valid(raise_exception=True)
+        print(serializer.validated_data)
+        plans = self.paginate_queryset(self.filter_queryset(self.get_queryset()))
+        for plan in plans:
+            plan.calculate_promotion(**serializer.validated_data)
+        serializer = self.get_serializer(plans, many=True)
+        return self.get_paginated_response(serializer.data)

apps/sample/admin.py

@@ -1,9 +0,0 @@
-from functools import update_wrapper
-
-from django.contrib import admin
-
-from .models import Sample
-
-
-admin.site.register(Sample)
-

apps/sample/models.py

@@ -1,21 +0,0 @@
-from django.conf import settings
-from django.db import models
-from oauth2_provider.settings import oauth2_settings
-
-from utils.models import BaseModel
-
-from django.utils.translation import gettext_lazy as _
-
-
-class SampleQuerySet(models.QuerySet):
-    pass
-
-
-class Sample(BaseModel):
-    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
-    application = models.ForeignKey(oauth2_settings.APPLICATION_MODEL, on_delete=models.PROTECT,
-                                    related_name='+', null=True, blank=True)
-
-    data = models.JSONField(null=True, blank=True)
-
-    objects = SampleQuerySet.as_manager()

apps/sample/serializers.py

@@ -1,14 +0,0 @@
-from rest_framework import serializers
-
-from .models import Sample
-
-
-class SampleSerializer(serializers.ModelSerializer):
-    class Meta:
-        model = Sample
-        fields = (
-            "user",
-            "application",
-            "data"
-        )
-        read_only_fields = ('user','application')

apps/sample/views.py

@@ -1,63 +0,0 @@
-from django.shortcuts import render
-
-from django.http import Http404
-from django.shortcuts import render
-from oauth2_provider.contrib.rest_framework import TokenMatchesOASRequirements
-from rest_framework import viewsets, mixins
-from rest_framework.generics import get_object_or_404
-from rest_framework.viewsets import GenericViewSet
-
-from .models import PushUser, PushMessage
-from .serializers import PushUserSerializer, PushMessageSerializer
-from apps.gooyal_oauth2.models import Application
-from apps.gooyal_oauth2.rest_framework import IsAuthenticatedOrTokenMatchesOASRequirements
-from apps.gooyal_oauth2.utils import get_application
-from ..users.models import User
-
-
-from django.shortcuts import render
-
-from django.http import Http404
-from django.shortcuts import render
-from django.urls import reverse
-from oauth2_provider.contrib.rest_framework import TokenMatchesOASRequirements
-from rest_framework import viewsets, mixins
-from rest_framework.generics import get_object_or_404
-from rest_framework.viewsets import GenericViewSet
-
-from .models import Sample
-from .serializers import SampleSerializer
-from apps.gooyal_oauth2.models import Application
-from apps.gooyal_oauth2.rest_framework import IsAuthenticatedOrTokenMatchesOASRequirements
-from apps.gooyal_oauth2.utils import get_application
-from ..users.models import User
-
-
-class ApplicationSampleViewSet(mixins.ListModelMixin,
-                      mixins.CreateModelMixin,
-                      GenericViewSet):
-    queryset = Sample.objects.all()
-    serializer_class = SampleSerializer
-
-    permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements]
-    required_alternate_scopes = {
-        "POST": [["service_template.application.user-sample:submit"]],
-        "GET": [["service_template.application.user-sample:list-retrieve"]],
-    }
-
-    def perform_create(self, serializer):
-        user_uuid = self.kwargs.get('user_uuid')
-        user = User.objects.get(uuid=user_uuid)
-        application = get_application(self.request)
-        instance = serializer.save(user=user, application=application)
-
-
-class UserSampleViewSet(mixins.RetrieveModelMixin,
-                          GenericViewSet):
-    queryset = Sample.objects.all()
-    serializer_class = SampleSerializer
-    permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements]
-    required_alternate_scopes = {
-        "GET": [["service_template.user.self-sample:list-retrieve"]],
-    }
-

main/settings.py

@@ -53,7 +53,7 @@ INSTALLED_APPS = [
     'utils',
     'apps.users',
     'apps.gooyal_oauth2',
-    # 'apps.sample',
+    'apps.promotions',
 ]
 
 MIDDLEWARE = [

main/urls.py

@@ -29,7 +29,7 @@ urlpatterns = [
     path('admin/', admin.site.urls),
     path('', include('utils.urls')),
     path('oauth2/', include('oauth2_provider.urls', namespace='oauth2_provider')),
-    # path('sample/', include('apps.sample.urls', namespace='sample')),
+    path('promotions/', include('apps.promotions.urls', namespace='promotions')),
 ]
 
 urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)