feat(notifications): add alert subscription api

Author: sudip-khanal

main/urls.py

@@ -153,6 +153,7 @@
 router.register(r"situation_report_type", api_views.SituationReportTypeViewset, basename="situation_report_type")
 router.register(r"subscription", notification_views.SubscriptionViewset, basename="subscription")
 router.register(r"surge_alert", notification_views.SurgeAlertViewset, basename="surge_alert")
+router.register(r"alert-subscription", notification_views.AlertSubscriptionViewSet, basename="alert_subscription")
 router.register(r"user", api_views.UserViewset, basename="user")
 router.register(r"flash-update", flash_views.FlashUpdateViewSet, basename="flash_update")
 router.register(r"flash-update-file", flash_views.FlashUpdateFileViewSet, basename="flash_update_file")

notifications/admin.py

@@ -66,3 +66,29 @@ def has_delete_permission(self, request, obj=None):
 admin.site.register(models.NotificationGUID, NotificationGUIDAdmin)
 admin.site.register(models.Subscription, SubscriptionAdmin)
 admin.site.register(models.SurgeAlert, SurgeAlertAdmin)
+
+
+@admin.register(models.HazardType)
+class AlertTypeAdmin(admin.ModelAdmin):
+    list_display = ("type",)
+
+
+@admin.register(models.AlertSubscription)
+class AlertSubscriptionAdmin(admin.ModelAdmin):
+    list_select_related = True
+    list_display = ("user", "created_at")
+    autocomplete_fields = ("user",)
+
+    def get_queryset(self, request):
+        return (
+            super()
+            .get_queryset(request)
+            .select_related(
+                "user",
+            )
+            .prefetch_related(
+                "countries",
+                "regions",
+                "hazard_types",
+            )
+        )

notifications/drf_views.py

@@ -1,17 +1,20 @@
 # from datetime import datetime, timedelta, timezone
 # from django.db.models import Q
+from django.db.models.query import QuerySet
 from django_filters import rest_framework as filters
 from django_filters.widgets import CSVWidget
-from rest_framework import viewsets
+from rest_framework import mixins, viewsets
 from rest_framework.authentication import TokenAuthentication
 from rest_framework.permissions import IsAuthenticated
 
 from deployments.models import MolnixTag
 from main.filters import CharInFilter
 from main.permissions import DenyGuestUserPermission
+from notifications.filter_set import AlertSubscriptionFilterSet
 
-from .models import Subscription, SurgeAlert
+from .models import AlertSubscription, Subscription, SurgeAlert
 from .serializers import (  # UnauthenticatedSurgeAlertSerializer,
+    AlertSubscriptionSerialize,
     SubscriptionSerializer,
     SurgeAlertCsvSerializer,
     SurgeAlertSerializer,
@@ -110,3 +113,34 @@ class SubscriptionViewset(viewsets.ModelViewSet):
 
     def get_queryset(self):
         return Subscription.objects.filter(user=self.request.user)
+
+
+class AlertSubscriptionViewSet(
+    viewsets.GenericViewSet,
+    mixins.ListModelMixin,
+    mixins.RetrieveModelMixin,
+    mixins.CreateModelMixin,
+    mixins.UpdateModelMixin,
+):
+    queryset = AlertSubscription.objects.all()
+    serializer_class = AlertSubscriptionSerialize
+    filterset_class = AlertSubscriptionFilterSet
+    lookup_field = "id"
+    permission_classes = [
+        IsAuthenticated,
+        DenyGuestUserPermission,
+    ]
+
+    def get_queryset(self) -> QuerySet[AlertSubscription]:
+        return (
+            super()
+            .get_queryset()
+            .select_related(
+                "user",
+            )
+            .prefetch_related(
+                "countries",
+                "regions",
+                "hazard_types",
+            )
+        )

notifications/enums.py

@@ -2,4 +2,7 @@
 
 enum_register = {
     "surge_alert_status": models.SurgeAlertStatus,
+    "alert_source": models.AlertSubscription.AlertSource,
+    "hazard_type": models.HazardType.Type,
+    "alert_per_day": models.AlertSubscription.AlertPerDay,
 }

notifications/factories.py

@@ -1,7 +1,9 @@
 import factory
 from factory import fuzzy
 
-from .models import SurgeAlert, SurgeAlertStatus
+from deployments.factories.user import UserFactory
+
+from .models import AlertSubscription, HazardType, SurgeAlert, SurgeAlertStatus
 
 
 class SurgeAlertFactory(factory.django.DjangoModelFactory):
@@ -21,3 +23,39 @@ def molnix_tags(self, create, extracted, **_):
         if extracted:
             for item in extracted:
                 self.molnix_tags.add(item)
+
+
+class AlertSubscriptionFactory(factory.django.DjangoModelFactory):
+    class Meta:
+        model = AlertSubscription
+
+    user = factory.SubFactory(UserFactory)
+
+    @factory.post_generation
+    def countries(self, create, extracted, **kwargs):
+        if not create:
+            return
+        if extracted:
+            for country in extracted:
+                self.countries.add(country)
+
+    @factory.post_generation
+    def regions(self, create, extracted, **kwargs):
+        if not create:
+            return
+        if extracted:
+            for region in extracted:
+                self.regions.add(region)
+
+    @factory.post_generation
+    def hazard_types(self, create, extracted, **kwargs):
+        if not create:
+            return
+        if extracted:
+            for alert_type in extracted:
+                self.hazard_types.add(alert_type)
+
+
+class HazardTypeFactory(factory.django.DjangoModelFactory):
+    class Meta:
+        model = HazardType

notifications/filter_set.py

@@ -0,0 +1,19 @@
+import django_filters as filters
+
+from api.models import Country, Region
+from notifications.models import AlertSubscription
+
+
+class AlertSubscriptionFilterSet(filters.FilterSet):
+    country = filters.ModelMultipleChoiceFilter(field_name="countries", queryset=Country.objects.all())
+    region = filters.ModelMultipleChoiceFilter(field_name="regions", queryset=Region.objects.all())
+    alert_source = filters.NumberFilter(field_name="alert_source", label="Alert Source")
+    hazard_type = filters.NumberFilter(field_name="hazard_types__type", label="Hazard Type")
+    alert_per_day = filters.ChoiceFilter(choices=AlertSubscription.AlertPerDay.choices, label="Alert Per Day")
+
+    class Meta:
+        model = AlertSubscription
+        fields = {
+            "countries__iso3": ("exact",),
+            "alert_per_day": ("exact",),
+        }

notifications/migrations/0016_hazardtype_alertsubscription.py

@@ -0,0 +1,47 @@
+# Generated by Django 4.2.19 on 2025-12-03 10:45
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('api', '0226_nsdinitiativescategory_and_more'),
+        ('notifications', '0015_rename_molnix_status_surgealert_molnix_status_old'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='HazardType',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('type', models.IntegerField(choices=[(100, 'Earthquake'), (200, 'Flood'), (300, 'Cyclone')], unique=True, verbose_name='Hazard Type')),
+            ],
+            options={
+                'verbose_name': 'Hazard Type',
+                'verbose_name_plural': 'Hazard Types',
+            },
+        ),
+        migrations.CreateModel(
+            name='AlertSubscription',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('alert_source', models.IntegerField(choices=[(100, 'Montandon')], default=100, verbose_name='Alert source')),
+                ('alert_per_day', models.IntegerField(choices=[(100, 'Five'), (200, 'Ten'), (300, 'Twenty'), (400, 'Fifty'), (500, 'Unlimited')], default=100, help_text='Maximum number of alerts sent to the user per day.', verbose_name='Alerts Per Day')),
+                ('created_at', models.DateTimeField(auto_now_add=True, verbose_name='Created At')),
+                ('updated_at', models.DateTimeField(auto_now=True, verbose_name='Updated At')),
+                ('countries', models.ManyToManyField(related_name='alert_subscriptions_by_country', to='api.country', verbose_name='Countries')),
+                ('hazard_types', models.ManyToManyField(help_text='Types of hazard the user is subscribed to.', related_name='+', to='notifications.hazardtype', verbose_name='Hazard Types')),
+                ('regions', models.ManyToManyField(blank=True, related_name='alert_subscriptions_by_region', to='api.region', verbose_name='Regions')),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='alert_subscriptions', to=settings.AUTH_USER_MODEL, verbose_name='User')),
+            ],
+            options={
+                'verbose_name': 'Alert Subscription',
+                'verbose_name_plural': 'Alert Subscriptions',
+                'ordering': ['-id'],
+            },
+        ),
+    ]

notifications/models.py

@@ -320,3 +320,97 @@ class NotificationGUID(models.Model):
     )
     email_type = models.CharField(max_length=600, null=True, blank=True)
     to_list = models.TextField(null=True, blank=True)
+
+
+class HazardType(models.Model):
+    """Model representing a hazard category."""
+
+    class Type(models.IntegerChoices):
+        EARTHQUAKE = 100, _("Earthquake")
+        FLOOD = 200, _("Flood")
+        CYCLONE = 300, _("Cyclone")
+
+    type = models.IntegerField(
+        choices=Type.choices,
+        unique=True,
+        verbose_name=_("Hazard Type"),
+    )
+
+    class Meta:
+        verbose_name = _("Hazard Type")
+        verbose_name_plural = _("Hazard Types")
+
+    def __str__(self):
+        return self.get_type_display()
+
+
+class AlertSubscription(models.Model):
+    class AlertSource(models.IntegerChoices):
+        MONTANDON = 100, _("Montandon")
+        """Alerts provided by the Montandon platform."""
+
+    class AlertPerDay(models.IntegerChoices):
+        """Enum representing the maximum number of alerts per day."""
+
+        FIVE = 100, _("Five")
+        """Receive up to 5 alerts per day."""
+
+        TEN = 200, _("Ten")
+        """Receive up to 10 alerts per day."""
+
+        TWENTY = 300, _("Twenty")
+        """Receive up to 20 alerts per day."""
+
+        FIFTY = 400, _("Fifty")
+        """Receive up to 50 alerts per day."""
+
+        UNLIMITED = 500, _("Unlimited")
+        """No daily alert limit."""
+
+    user = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        verbose_name=_("User"),
+        on_delete=models.CASCADE,
+        related_name="alert_subscriptions",
+    )
+    countries = models.ManyToManyField(
+        Country,
+        related_name="alert_subscriptions_by_country",
+        verbose_name=_("Countries"),
+    )
+    regions = models.ManyToManyField(
+        Region,
+        related_name="alert_subscriptions_by_region",
+        blank=True,
+        verbose_name=_("Regions"),
+    )
+    alert_source = models.IntegerField(
+        choices=AlertSource.choices,
+        default=AlertSource.MONTANDON,
+        verbose_name=_("Alert source"),
+    )
+
+    hazard_types = models.ManyToManyField(
+        HazardType,
+        related_name="+",
+        verbose_name=_("Hazard Types"),
+        help_text=_("Types of hazard the user is subscribed to."),
+    )
+    alert_per_day = models.IntegerField(
+        choices=AlertPerDay.choices,
+        default=AlertPerDay.FIVE,
+        verbose_name=_("Alerts Per Day"),
+        help_text=_("Maximum number of alerts sent to the user per day."),
+    )
+    created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Created At"))
+    updated_at = models.DateTimeField(auto_now=True, verbose_name=_("Updated At"))
+    # Typing
+    id = int
+
+    class Meta:
+        ordering = ["-id"]
+        verbose_name = _("Alert Subscription")
+        verbose_name_plural = _("Alert Subscriptions")
+
+    def __str__(self):
+        return f"Alert subscription for {self.user}"

notifications/serializers.py

@@ -3,12 +3,14 @@
 from api.serializers import (
     MiniCountrySerializer,
     MiniEventSerializer,
+    MiniRegionSerialzier,
     SurgeEventSerializer,
+    UserNameSerializer,
 )
 from deployments.serializers import MolnixTagSerializer
 from lang.serializers import ModelSerializer
 
-from .models import Subscription, SurgeAlert
+from .models import AlertSubscription, HazardType, Subscription, SurgeAlert
 
 
 class SurgeAlertSerializer(ModelSerializer):
@@ -258,3 +260,42 @@ class Meta:
             "rtype",
             "rtype_display",
         )
+
+
+class HazardTypeSerializer(ModelSerializer):
+
+    type_display = serializers.CharField(source="get_alert_type_display", read_only=True)
+
+    class Meta:
+        model = HazardType
+        fields = (
+            "id",
+            "type",
+            "type_display",
+        )
+
+
+class AlertSubscriptionSerialize(ModelSerializer):
+    user_detail = UserNameSerializer(source="user", read_only=True)
+    countries_detail = MiniCountrySerializer(source="countries", many=True, read_only=True)
+    regions_detail = MiniRegionSerialzier(source="regions", many=True, read_only=True)
+    hazard_types_detail = HazardTypeSerializer(source="hazard_types", many=True, read_only=True)
+    alert_per_day_display = serializers.CharField(source="get_alert_per_day_display", read_only=True)
+
+    class Meta:
+        model = AlertSubscription
+        fields = (
+            "id",
+            "user",
+            "countries",
+            "regions",
+            "hazard_types",
+            "alert_per_day",
+            "user_detail",
+            "countries_detail",
+            "regions_detail",
+            "hazard_types_detail",
+            "alert_per_day_display",
+            "created_at",
+            "updated_at",
+        )