feat: Added helper methods NotificationType (#4)

Added helper methods to NotificationType:

- set_email_frequency
- get_email_frequency
- reset_email_frequency_to_default
- get_enabled_channels
- is_channel_enabled
- disable_channel
- enable_channel

Also internally everything now uses types rather than instances, and no more string arguments either.

Author: kevinrenskers

README.md

@@ -116,18 +116,10 @@ from generic_notifications.frequencies import RealtimeFrequency
 from myapp.notifications import CommentNotification
 
 # Disable email channel for comment notifications
-DisabledNotificationTypeChannel.objects.create(
-    user=user,
-    notification_type=CommentNotification.key,
-    channel=EmailChannel.key
-)
+CommentNotification.disable_channel(user=user, channel=EmailChannel)
 
 # Change to realtime digest for a notification type
-EmailFrequency.objects.update_or_create(
-    user=user,
-    notification_type=CommentNotification.key,
-    defaults={'frequency': RealtimeFrequency.key}
-)
+CommentNotification.set_email_frequency(user=user, frequency=RealtimeFrequency)
 ```
 
 This project doesn't come with a UI (view + template) for managing user preferences, but an example is provided in the [example app](#example-app).

generic_notifications/__init__.py

@@ -56,15 +56,10 @@ def send_notification(
             )
 
     # Determine which channels are enabled for this user/notification type
-    enabled_channels = []
-    enabled_channel_instances = []
-    for channel_instance in registry.get_all_channels():
-        if channel_instance.is_enabled(recipient, notification_type.key):
-            enabled_channels.append(channel_instance.key)
-            enabled_channel_instances.append(channel_instance)
+    enabled_channel_classes = notification_type.get_enabled_channels(recipient)
 
     # Don't create notification if no channels are enabled
-    if not enabled_channels:
+    if not enabled_channel_classes:
         return None
 
     # Create the notification record with enabled channels
@@ -73,7 +68,7 @@ def send_notification(
         notification_type=notification_type.key,
         actor=actor,
         target=target,
-        channels=enabled_channels,
+        channels=[channel_cls.key for channel_cls in enabled_channel_classes],
         subject=subject,
         text=text,
         url=url,
@@ -87,6 +82,7 @@ def send_notification(
             notification.save()
 
             # Process through enabled channels only
+            enabled_channel_instances = [channel_cls() for channel_cls in enabled_channel_classes]
             for channel_instance in enabled_channel_instances:
                 try:
                     channel_instance.process(notification)

generic_notifications/channels.py

@@ -8,7 +8,7 @@
 from django.template.loader import render_to_string
 from django.utils import timezone
 
-from .frequencies import DailyFrequency, NotificationFrequency
+from .frequencies import NotificationFrequency
 from .registry import registry
 
 if TYPE_CHECKING:
@@ -33,23 +33,6 @@ def process(self, notification: "Notification") -> None:
         """
         pass
 
-    def is_enabled(self, user: Any, notification_type: str) -> bool:
-        """
-        Check if user has this channel enabled for this notification type.
-
-        Args:
-            user: User instance
-            notification_type: Notification type key
-
-        Returns:
-            bool: True if enabled (default), False if disabled
-        """
-        from .models import DisabledNotificationTypeChannel
-
-        return not DisabledNotificationTypeChannel.objects.filter(
-            user=user, notification_type=notification_type, channel=self.key
-        ).exists()
-
 
 def register(cls: Type[NotificationChannel]) -> Type[NotificationChannel]:
     """
@@ -106,37 +89,14 @@ def process(self, notification: "Notification") -> None:
         Args:
             notification: Notification instance to process
         """
-        frequency = self.get_frequency(notification.recipient, notification.notification_type)
+        # Get notification type class from key
+        notification_type_cls = registry.get_type(notification.notification_type)
+        frequency_cls = notification_type_cls.get_email_frequency(notification.recipient)
 
         # Send immediately if realtime, otherwise leave for digest
-        if frequency and frequency.is_realtime:
+        if frequency_cls and frequency_cls.is_realtime:
             self.send_email_now(notification)
 
-    def get_frequency(self, user: Any, notification_type: str) -> NotificationFrequency:
-        """
-        Get the user's email frequency preference for this notification type.
-
-        Args:
-            user: User instance
-            notification_type: Notification type key
-
-        Returns:
-            NotificationFrequency: NotificationFrequency instance (defaults to notification type's default)
-        """
-        from .models import EmailFrequency
-
-        try:
-            email_frequency = EmailFrequency.objects.get(user=user, notification_type=notification_type)
-            return registry.get_frequency(email_frequency.frequency)
-        except (EmailFrequency.DoesNotExist, KeyError):
-            # Get the notification type's default frequency
-            try:
-                notification_type_obj = registry.get_type(notification_type)
-                return notification_type_obj.default_email_frequency()
-            except (KeyError, AttributeError):
-                # Fallback to realtime if notification type not found or no default
-                return DailyFrequency()
-
     def send_email_now(self, notification: "Notification") -> None:
         """
         Send an individual email notification immediately.
@@ -196,7 +156,7 @@ def send_email_now(self, notification: "Notification") -> None:
 
     @classmethod
     def send_digest_emails(
-        cls, user: Any, notifications: "QuerySet[Notification]", frequency: NotificationFrequency | None = None
+        cls, user: Any, notifications: "QuerySet[Notification]", frequency: type[NotificationFrequency] | None = None
     ):
         """
         Send a digest email to a specific user with specific notifications.
@@ -207,14 +167,12 @@ def send_digest_emails(
             notifications: QuerySet of notifications to include in digest
             frequency: The frequency for template context
         """
-        from .models import Notification
-
         if not notifications.exists():
             return
 
         try:
             # Group notifications by type for better digest formatting
-            notifications_by_type: dict[str, list[Notification]] = {}
+            notifications_by_type: dict[str, list["Notification"]] = {}
             for notification in notifications:
                 if notification.notification_type not in notifications_by_type:
                     notifications_by_type[notification.notification_type] = []

generic_notifications/management/commands/send_digest_emails.py

@@ -1,11 +1,14 @@
 import logging
 
 from django.contrib.auth import get_user_model
+from django.contrib.auth.models import AbstractUser
 from django.core.management.base import BaseCommand
 
 from generic_notifications.channels import EmailChannel
+from generic_notifications.frequencies import NotificationFrequency
 from generic_notifications.models import Notification
 from generic_notifications.registry import registry
+from generic_notifications.types import NotificationType
 
 User = get_user_model()
 
@@ -48,23 +51,22 @@ def handle(self, *args, **options):
             return
 
         # Setup
-        email_channel = EmailChannel()
         all_notification_types = registry.get_all_types()
 
         # Get the specific frequency (required argument)
         try:
-            frequency = registry.get_frequency(target_frequency)
+            frequency_cls = registry.get_frequency(target_frequency)
         except KeyError:
             logger.error(f"Frequency '{target_frequency}' not found")
             return
 
-        if frequency.is_realtime:
+        if frequency_cls.is_realtime:
             logger.error(f"Frequency '{target_frequency}' is realtime, not a digest frequency")
             return
 
         total_emails_sent = 0
 
-        logger.info(f"Processing {frequency.name} digests...")
+        logger.info(f"Processing {frequency_cls.name} digests...")
 
         # Find all users who have unsent, unread notifications for email channel
         users_with_notifications = User.objects.filter(
@@ -75,31 +77,29 @@ def handle(self, *args, **options):
 
         for user in users_with_notifications:
             # Determine which notification types should use this frequency for this user
-            relevant_types = self.get_notification_types_for_frequency(
-                user,
-                frequency.key,
-                all_notification_types,
-                email_channel,
-            )
+            relevant_types = self.get_notification_types_for_frequency(user, frequency_cls, all_notification_types)
 
             if not relevant_types:
                 continue
 
             # Get unsent notifications for these types
             # Exclude read notifications - don't email what user already saw on website
+            relevant_type_keys = [nt.key for nt in relevant_types]
             notifications = Notification.objects.filter(
                 recipient=user,
-                notification_type__in=relevant_types,
+                notification_type__in=relevant_type_keys,
                 email_sent_at__isnull=True,
                 read__isnull=True,
                 channels__icontains=f'"{EmailChannel.key}"',
             ).order_by("-added")
 
             if notifications.exists():
-                logger.info(f"  User {user.email}: {notifications.count()} notifications for {frequency.name} digest")
+                logger.info(
+                    f"  User {user.email}: {notifications.count()} notifications for {frequency_cls.name} digest"
+                )
 
                 if not dry_run:
-                    EmailChannel.send_digest_emails(user, notifications, frequency)
+                    EmailChannel.send_digest_emails(user, notifications, frequency_cls)
 
                 total_emails_sent += 1
 
@@ -117,18 +117,30 @@ def handle(self, *args, **options):
         else:
             logger.info(f"Successfully sent {total_emails_sent} digest emails")
 
-    def get_notification_types_for_frequency(self, user, frequency_key, all_notification_types, email_channel):
+    def get_notification_types_for_frequency(
+        self,
+        user: AbstractUser,
+        wanted_frequency: type[NotificationFrequency],
+        all_notification_types: list[type["NotificationType"]],
+    ) -> list[type["NotificationType"]]:
         """
         Get all notification types that should use this frequency for the given user.
         This includes both explicit preferences and types that default to this frequency.
         Since notifications are only created for enabled channels, we don't need to check is_enabled.
+
+        Args:
+            user: The user to check preferences for
+            wanted_frequency: The frequency to filter by (e.g. DailyFrequency, RealtimeFrequency)
+            all_notification_types: List of all registered notification type classes
+
+        Returns:
+            List of notification type classes that use this frequency for this user
         """
-        relevant_types = set()
+        relevant_types: list[type["NotificationType"]] = []
 
         for notification_type in all_notification_types:
-            # Use EmailChannel's get_frequency method to get the frequency for this user/type
-            user_frequency = email_channel.get_frequency(user, notification_type.key)
-            if user_frequency.key == frequency_key:
-                relevant_types.add(notification_type.key)
+            user_frequency = notification_type.get_email_frequency(user)
+            if user_frequency.key == wanted_frequency.key:
+                relevant_types.append(notification_type)
 
-        return list(relevant_types)
+        return relevant_types

generic_notifications/models.py

@@ -43,7 +43,7 @@ class Meta:
 
     def clean(self):
         try:
-            notification_type_obj = registry.get_type(self.notification_type)
+            notification_type_cls = registry.get_type(self.notification_type)
         except KeyError:
             available_types = [t.key for t in registry.get_all_types()]
             if available_types:
@@ -56,10 +56,10 @@ def clean(self):
                 )
 
         # Check if trying to disable a required channel
-        required_channel_keys = [cls.key for cls in notification_type_obj.required_channels]
+        required_channel_keys = [cls.key for cls in notification_type_cls.required_channels]
         if self.channel in required_channel_keys:
             raise ValidationError(
-                f"Cannot disable {self.channel} channel for {notification_type_obj.name} - this channel is required"
+                f"Cannot disable {self.channel} channel for {notification_type_cls.name} - this channel is required"
             )
 
         try:
@@ -204,7 +204,8 @@ def get_subject(self) -> str:
 
         # Get the notification type and use its dynamic generation
         try:
-            notification_type = registry.get_type(self.notification_type)
+            notification_type_cls = registry.get_type(self.notification_type)
+            notification_type = notification_type_cls()
             return notification_type.get_subject(self) or notification_type.description
         except KeyError:
             return f"Notification: {self.notification_type}"
@@ -216,7 +217,8 @@ def get_text(self) -> str:
 
         # Get the notification type and use its dynamic generation
         try:
-            notification_type = registry.get_type(self.notification_type)
+            notification_type_cls = registry.get_type(self.notification_type)
+            notification_type = notification_type_cls()
             return notification_type.get_text(self)
         except KeyError:
             return "You have a new notification"

generic_notifications/preferences.py

@@ -1,11 +1,10 @@
-from typing import TYPE_CHECKING, Any, Dict, List
+from typing import Any, Dict, List
+
+from django.contrib.auth.models import AbstractUser
 
 from .models import DisabledNotificationTypeChannel, EmailFrequency
 from .registry import registry
 
-if TYPE_CHECKING:
-    from django.contrib.auth.models import AbstractUser
-
 
 def get_notification_preferences(user: "AbstractUser") -> List[Dict[str, Any]]:
     """
@@ -90,16 +89,15 @@ def save_notification_preferences(user: "AbstractUser", form_data: Dict[str, Any
 
             # If checkbox not checked, create disabled entry
             if form_key not in form_data:
-                DisabledNotificationTypeChannel.objects.create(
-                    user=user, notification_type=type_key, channel=channel_key
-                )
+                notification_type.disable_channel(user=user, channel=channel)
 
         # Handle email frequency preference
         if "email" in [ch.key for ch in channels.values()]:
             frequency_key = f"{type_key}__frequency"
             if frequency_key in form_data:
                 frequency_value = form_data[frequency_key]
                 if frequency_value in frequencies:
+                    frequency_obj = frequencies[frequency_value]
                     # Only save if different from default
                     if frequency_value != notification_type.default_email_frequency.key:
-                        EmailFrequency.objects.create(user=user, notification_type=type_key, frequency=frequency_value)
+                        notification_type.set_email_frequency(user=user, frequency=frequency_obj)