diff --git a/openedx/core/djangoapps/notifications/base_notification.py b/openedx/core/djangoapps/notifications/base_notification.py index de39b609a9e9..7b35a0ce61dd 100644 --- a/openedx/core/djangoapps/notifications/base_notification.py +++ b/openedx/core/djangoapps/notifications/base_notification.py @@ -25,7 +25,7 @@ class NotificationType(TypedDict): notification_app: str # Unique identifier for this notification type. name: str - # Mark this as a core notification. + # Whether this notification type uses the notification app's default settings. # When True, user preferences are taken from the notification app's configuration, # overriding the `web`, `email`, `push`, `email_cadence`, and `non_editable` attributes set here. use_app_defaults: bool @@ -38,8 +38,7 @@ class NotificationType(TypedDict): content_context: dict[str, Any] filters: list[str] - # All fields below are required unless `is_core` is True. - # Core notifications take this config from the associated notification app instead (and ignore anything set here). + # All fields below are required unless `use_app_defaults` is True. # Set to True to enable delivery on web. web: NotRequired[bool] @@ -306,9 +305,7 @@ class NotificationApp(TypedDict): Each notification type defined in COURSE_NOTIFICATION_TYPES also references an app. - Each notification type can also be optionally defined as a core notification. In this case, the delivery preferences for that notification are taken - from the `core_*` fields of the associated notification app. """ # Set to True to enable this app and linked notification types. enabled: bool @@ -366,106 +363,6 @@ class NotificationApp(TypedDict): COURSE_NOTIFICATION_APPS = get_notification_apps_config() -class NotificationTypeManager: - """ - Manager for notification types - """ - - def __init__(self): - self.notification_types = COURSE_NOTIFICATION_TYPES - - def get_notification_types_by_app(self, notification_app: str): - """ - Returns notification types for the given notification app name. - """ - return [ - notification_type.copy() for _, notification_type in self.notification_types.items() - if notification_type.get('notification_app', None) == notification_app - ] - - def get_core_and_non_core_notification_types( - self, notification_app: str - ) -> tuple[NotificationType, NotificationType]: - """ - Returns notification types for the given app name, split by core and non core. - - Return type is a tuple of (core_notification_types, non_core_notification_types). - """ - notification_types = self.get_notification_types_by_app(notification_app) - core_notification_types = [] - non_core_notification_types = [] - for notification_type in notification_types: - if notification_type.get('use_app_defaults', None): - core_notification_types.append(notification_type) - else: - non_core_notification_types.append(notification_type) - return core_notification_types, non_core_notification_types - - @staticmethod - def get_non_core_notification_type_preferences(non_core_notification_types, email_opt_out=False): - """ - Returns non-core notification type preferences for the given notification types. - """ - non_core_notification_type_preferences = {} - for notification_type in non_core_notification_types: - non_core_notification_type_preferences[notification_type.get('name')] = { - 'web': notification_type.get('web', False), - 'email': False if email_opt_out else notification_type.get('email', False), - 'push': notification_type.get('push', False), - 'email_cadence': notification_type.get('email_cadence', 'Daily'), - } - return non_core_notification_type_preferences - - def get_notification_app_preference(self, notification_app, email_opt_out=False): - """ - Returns notification app preferences for the given notification app. - """ - core_notification_types, non_core_notification_types = self.get_core_and_non_core_notification_types( - notification_app, - ) - non_core_notification_types_preferences = self.get_non_core_notification_type_preferences( - non_core_notification_types, email_opt_out - ) - core_notification_types_name = [notification_type.get('name') for notification_type in core_notification_types] - return non_core_notification_types_preferences, core_notification_types_name - - -class NotificationAppManager: - """ - Notification app manager - """ - - def add_core_notification_preference(self, notification_app_attrs, notification_types, email_opt_out=False): - """ - Adds core notification preference for the given notification app. - """ - notification_types['core'] = { - 'web': notification_app_attrs.get('web', False), - 'email': False if email_opt_out else notification_app_attrs.get('email', False), - 'push': notification_app_attrs.get('push', False), - 'email_cadence': notification_app_attrs.get('email_cadence', 'Daily'), - } - - def get_notification_app_preferences(self, email_opt_out=False): - """ - Returns notification app preferences for the given name. - """ - course_notification_preference_config = {} - for notification_app_key, notification_app_attrs in COURSE_NOTIFICATION_APPS.items(): - notification_app_preferences = {} - notification_types, core_notifications = NotificationTypeManager().get_notification_app_preference( - notification_app_key, - email_opt_out - ) - self.add_core_notification_preference(notification_app_attrs, notification_types, email_opt_out) - - notification_app_preferences['enabled'] = notification_app_attrs.get('enabled', False) - notification_app_preferences['core_notification_types'] = core_notifications - notification_app_preferences['notification_types'] = notification_types - course_notification_preference_config[notification_app_key] = notification_app_preferences - return course_notification_preference_config - - def get_notification_content(notification_type: str, context: dict[str, Any]): """ Returns notification content for the given notification type with provided context. @@ -489,8 +386,8 @@ def get_notification_content(notification_type: str, context: dict[str, Any]): if notification_type == 'course_update': notification_type = 'course_updates' - # Retrieve the notification type object from NotificationTypeManager. - notification_type = NotificationTypeManager().notification_types.get(notification_type, None) + # Retrieve the notification type object from the default preferences (derived from COURSE_NOTIFICATION_TYPES). + notification_type = get_default_values_of_preferences().get(notification_type, None) if notification_type: # Check if the notification is grouped. @@ -512,19 +409,6 @@ def get_notification_content(notification_type: str, context: dict[str, Any]): return '' -def get_default_values_of_preference(notification_app, notification_type): - """ - Returns default preference for notification_type - """ - default_prefs = NotificationAppManager().get_notification_app_preferences() - app_prefs = default_prefs.get(notification_app, {}) - core_notification_types = app_prefs.get('core_notification_types', []) - notification_types = app_prefs.get('notification_types', {}) - if notification_type in core_notification_types: - return notification_types.get('core', {}) - return notification_types.get(notification_type, {}) - - def get_default_values_of_preferences() -> dict[str, dict[str, Any]]: """ Returns default preferences for all notification apps diff --git a/openedx/core/djangoapps/notifications/serializers.py b/openedx/core/djangoapps/notifications/serializers.py index ce2fccb926b4..7dc1e24f3c5c 100644 --- a/openedx/core/djangoapps/notifications/serializers.py +++ b/openedx/core/djangoapps/notifications/serializers.py @@ -138,17 +138,6 @@ def validate_notification_channel(notification_channel: str) -> str: return notification_channel -def get_non_editable_channels(app_name): - """ - Returns a dict of notification: [non-editable channels] for the given app name. - """ - non_editable = {"core": COURSE_NOTIFICATION_APPS[app_name].get("non_editable", [])} - for type_name, type_dict in COURSE_NOTIFICATION_TYPES.items(): - if type_dict.get("non_editable") and not type_dict["is_core"]: - non_editable[type_name] = type_dict["non_editable"] - return non_editable - - def add_non_editable_in_preference(preference): """ Add non_editable preferences to the preference dict @@ -211,7 +200,6 @@ def validate(self, attrs): # Validate notification type if all([ not COURSE_NOTIFICATION_TYPES.get(notification_type), - notification_type != "core", notification_type != "grouped_notification", ]): raise ValidationError(f'{notification_type} is not a valid notification type.') diff --git a/openedx/core/djangoapps/notifications/tasks.py b/openedx/core/djangoapps/notifications/tasks.py index f44e4ce19729..00479da9c884 100644 --- a/openedx/core/djangoapps/notifications/tasks.py +++ b/openedx/core/djangoapps/notifications/tasks.py @@ -15,8 +15,7 @@ from openedx.core.djangoapps.notifications.audience_filters import NotificationFilter from openedx.core.djangoapps.notifications.base_notification import ( COURSE_NOTIFICATION_TYPES, - get_default_values_of_preference, - get_notification_content + get_notification_content, get_default_values_of_preferences ) from openedx.core.djangoapps.notifications.email.tasks import send_immediate_cadence_email @@ -120,7 +119,7 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c grouping_enabled = group_by_id and grouping_function is not None generated_notification = None sender_id = context.pop('sender_id', None) - default_web_config = get_default_values_of_preference(app_name, notification_type).get('web', False) + default_web_config = get_default_values_of_preferences().get(notification_type, {}).get('web', False) generated_notification_audience = [] email_notification_mapping = {} push_notification_audience = [] diff --git a/openedx/core/djangoapps/notifications/tests/test_views.py b/openedx/core/djangoapps/notifications/tests/test_views.py index edb72a629e90..36c0cd21ba15 100644 --- a/openedx/core/djangoapps/notifications/tests/test_views.py +++ b/openedx/core/djangoapps/notifications/tests/test_views.py @@ -31,14 +31,12 @@ Notification, NotificationPreference ) -from openedx.core.djangoapps.notifications.serializers import ( - add_info_to_notification_config, - add_non_editable_in_preference -) +from openedx.core.djangoapps.notifications.serializers import add_non_editable_in_preference + from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase from xmodule.modulestore.tests.factories import CourseFactory -from ..base_notification import COURSE_NOTIFICATION_APPS, NotificationTypeManager, COURSE_NOTIFICATION_TYPES, \ +from ..base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES, \ get_default_values_of_preferences from ..utils import get_notification_types_with_visibility_settings, exclude_inaccessible_preferences @@ -560,377 +558,6 @@ def remove_notifications_with_visibility_settings(expected_response): return expected_response -@ddt.ddt -class TestNotificationPreferencesView(ModuleStoreTestCase): - """ - Tests for the NotificationPreferencesView API view. - """ - - def setUp(self): - # Set up a user and API client - super().setUp() - self.default_data = { - "status": "success", - "message": "Notification preferences retrieved successfully.", - "data": { - "discussion": { - "enabled": True, - "core_notification_types": [ - "new_comment_on_response", - "new_comment", - "new_response", - "response_on_followed_post", - "comment_on_followed_post", - "response_endorsed_on_thread", - "response_endorsed" - ], - "notification_types": { - "new_discussion_post": { - "web": False, - "email": False, - "push": False, - "email_cadence": "Daily" - }, - "new_question_post": { - "web": False, - "email": False, - "push": False, - "email_cadence": "Daily" - }, - "content_reported": { - "web": True, - "email": True, - "push": False, - "email_cadence": "Daily" - }, - "new_instructor_all_learners_post": { - "web": True, - "email": True, - "push": False, - "email_cadence": "Daily" - }, - "core": { - "web": True, - "email": True, - "push": True, - "email_cadence": "Daily" - } - }, - "non_editable": { - "new_discussion_post": ["push"], - "new_question_post": ["push"], - "content_reported": ["push"], - "new_instructor_all_learners_post": ["push"] - } - }, - "updates": { - "enabled": True, - "core_notification_types": [], - "notification_types": { - "course_updates": { - "web": True, - "email": True, - "push": False, - "email_cadence": "Daily" - }, - "core": { - "web": True, - "email": True, - "push": True, - "email_cadence": "Daily" - } - }, - "non_editable": { - "course_updates": ["push"], - } - }, - "grading": { - "enabled": True, - "core_notification_types": [], - "notification_types": { - "ora_staff_notifications": { - "web": True, - "email": False, - "push": False, - "email_cadence": "Daily" - }, - "ora_grade_assigned": { - "web": True, - "email": True, - "push": False, - "email_cadence": "Daily" - }, - "core": { - "web": True, - "email": True, - "push": True, - "email_cadence": "Daily" - } - }, - "non_editable": { - "ora_grade_assigned": ["push"], - "ora_staff_notifications": ["push"] - } - }, - } - } - self.TEST_PASSWORD = 'testpass' - self.user = UserFactory(password=self.TEST_PASSWORD) - self.client = APIClient() - self.client.force_authenticate(user=self.user) - self.url = reverse('notification-preferences-aggregated-v2') # Adjust with the actual name - self.course = CourseFactory.create(display_name='test course 1', run="Testing_course_1") - - @ddt.data( - ("forum", FORUM_ROLE_ADMINISTRATOR, ['content_reported'], ['ora_staff_notifications']), - ("forum", FORUM_ROLE_MODERATOR, ['content_reported'], ['ora_staff_notifications']), - ("forum", FORUM_ROLE_COMMUNITY_TA, ['content_reported'], ['ora_staff_notifications']), - ("course", CourseStaffRole.ROLE, ['ora_staff_notifications'], ['content_reported']), - ("course", CourseInstructorRole.ROLE, ['ora_staff_notifications'], ['content_reported']), - (None, None, [], ['ora_staff_notifications', 'content_reported']), - ) - @ddt.unpack - def test_get_notification_preferences(self, role_type, role, visible_apps, hidden_apps): - """ - Test: Notification preferences visibility for users with forum, course, or no role. - """ - role_instance = None - - if role_type == "course": - if role == CourseInstructorRole.ROLE: - CourseStaffRole(self.course.id).add_users(self.user) - else: - CourseInstructorRole(self.course.id).add_users(self.user) - self.client.login(username=self.user.username, password='testpass') - - elif role_type == "forum": - role_instance = RoleFactory(name=role, course_id=self.course.id) - role_instance.users.add(self.user) - - response = self.client.get(self.url) - - self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual(response.data['status'], 'success') - self.assertIn('data', response.data) - - expected_data = exclude_inaccessible_preferences(self.default_data['data'], self.user) - expected_data = add_non_editable_in_preference(expected_data) - expected_data = add_info_to_notification_config(expected_data) - - self.assertEqual(response.data['data'], expected_data) - - notification_apps = {} - for app in ['discussion', 'grading']: - notification_apps.update(response.data['data'][app]['notification_types']) - - for app in visible_apps: - self.assertIn(app, notification_apps, msg=f"{app} should be visible for role: {role_type}") - - for app in hidden_apps: - self.assertNotIn(app, notification_apps, msg=f"{app} should NOT be visible for role: {role_type}") - - if role_type == "forum": - role_instance.users.clear() - elif role_type == "course": - if role == CourseInstructorRole.ROLE: - CourseStaffRole(self.course.id).remove_users(self.user) - else: - CourseInstructorRole(self.course.id).remove_users(self.user) - - def test_if_data_is_correctly_aggregated(self): - """ - Test case: Check if the data is correctly formatted - """ - - self.client.get(self.url) - NotificationPreference.objects.all().update( - web=False, - push=False, - email=False, - ) - response = self.client.get(self.url) - self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual(response.data['status'], 'success') - self.assertIn('data', response.data) - data = { - "status": "success", - "show_preferences": False, - "message": "Notification preferences retrieved successfully.", - "data": { - "discussion": { - "enabled": True, - "core_notification_types": [ - "new_comment_on_response", - "new_comment", - "new_response", - "response_on_followed_post", - "comment_on_followed_post", - "response_endorsed_on_thread", - "response_endorsed" - ], - "notification_types": { - "new_discussion_post": { - "web": False, - "email": False, - "push": False, - "email_cadence": "Daily", - "info": "" - }, - "new_question_post": { - "web": False, - "email": False, - "push": False, - "email_cadence": "Daily", - "info": "" - }, - "new_instructor_all_learners_post": { - "web": False, - "email": False, - "push": False, - "email_cadence": "Daily", - "info": "" - }, - "core": { - "web": False, - "email": False, - "push": False, - "email_cadence": "Daily", - "info": "Notifications for responses and comments on your posts, and the ones you’re " - "following, including endorsements to your responses and on your posts." - } - }, - "non_editable": { - "new_discussion_post": ["push"], - "new_question_post": ["push"], - "new_instructor_all_learners_post": ["push"] - } - }, - "updates": { - "enabled": True, - "core_notification_types": [], - "notification_types": { - "course_updates": { - "web": False, - "email": False, - "push": False, - "email_cadence": "Daily", - "info": "" - }, - "core": { - "web": True, - "email": True, - "push": True, - "email_cadence": "Daily", - "info": "Notifications for new announcements and updates from the course team." - } - }, - "non_editable": { - "course_updates": ["push"], - } - }, - "grading": { - "enabled": True, - "core_notification_types": [], - "notification_types": { - "ora_grade_assigned": { - "web": False, - "email": False, - "push": False, - "email_cadence": "Daily", - "info": "" - }, - "core": { - "web": True, - "email": True, - "push": True, - "email_cadence": "Daily", - "info": "Notifications for submission grading." - } - }, - "non_editable": { - "ora_grade_assigned": ["push"] - } - }, - } - } - self.assertEqual(response.data, data) - - def test_api_view_permissions(self): - """ - Test case: Ensure the API view has the correct permissions - """ - # Check if the view requires authentication - self.client.logout() - response = self.client.get(self.url) - self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) - - # Re-authenticate and check again - self.client.force_authenticate(user=self.user) - response = self.client.get(self.url) - self.assertEqual(response.status_code, status.HTTP_200_OK) - - def test_update_preferences_core(self): - """ - Test case: Update notification preferences for the authenticated user - """ - update_data = { - "notification_app": "discussion", - "notification_type": "core", - "notification_channel": "email_cadence", - "email_cadence": "Weekly" - } - __, core_types = NotificationTypeManager().get_notification_app_preference('discussion') - self.client.get(self.url) - response = self.client.put(self.url, update_data, format='json') - self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual(response.data['status'], 'success') - cadence_set = NotificationPreference.objects.filter(user=self.user, type__in=core_types).values_list( - 'email_cadence', flat=True - ) - self.assertEqual(len(set(cadence_set)), 1) - self.assertIn('Weekly', set(cadence_set)) - - def test_update_preferences(self): - """ - Test case: Update notification preferences for the authenticated user - """ - update_data = { - "notification_app": "discussion", - "notification_type": "new_discussion_post", - "notification_channel": "web", - "value": True - } - self.client.get(self.url) - response = self.client.put(self.url, update_data, format='json') - self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual(response.data['status'], 'success') - preference = NotificationPreference.objects.get( - type='new_discussion_post', - user__id=self.user.id - ) - self.assertEqual(preference.web, True) - - def test_update_preferences_non_core_email(self): - """ - Test case: Update notification preferences for the authenticated user - """ - update_data = { - "notification_app": "discussion", - "notification_type": "new_discussion_post", - "notification_channel": "email_cadence", - "email_cadence": 'Weekly' - } - self.client.get(self.url) - response = self.client.put(self.url, update_data, format='json') - self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual(response.data['status'], 'success') - preference = NotificationPreference.objects.get( - type='new_discussion_post', - user__id=self.user.id - ) - self.assertEqual(preference.email_cadence, 'Weekly') - - @ddt.ddt class TestNotificationPreferencesViewV3(ModuleStoreTestCase): """ diff --git a/openedx/core/djangoapps/notifications/urls.py b/openedx/core/djangoapps/notifications/urls.py index 279f02a97b60..2ac80440f109 100644 --- a/openedx/core/djangoapps/notifications/urls.py +++ b/openedx/core/djangoapps/notifications/urls.py @@ -10,18 +10,12 @@ NotificationListAPIView, NotificationReadAPIView, preference_update_from_encrypted_username_view, - NotificationPreferencesView, NotificationPreferencesViewV3, ) router = routers.DefaultRouter() urlpatterns = [ - path( - 'v2/configurations/', - NotificationPreferencesView.as_view(), - name='notification-preferences-aggregated-v2' - ), path( 'v3/configurations/', NotificationPreferencesViewV3.as_view(), diff --git a/openedx/core/djangoapps/notifications/views.py b/openedx/core/djangoapps/notifications/views.py index 57d720e30124..bb419e46ecc8 100644 --- a/openedx/core/djangoapps/notifications/views.py +++ b/openedx/core/djangoapps/notifications/views.py @@ -19,8 +19,7 @@ from openedx.core.djangoapps.notifications.models import NotificationPreference from openedx.core.djangoapps.notifications.permissions import allow_any_authenticated_user -from .base_notification import COURSE_NOTIFICATION_APPS, NotificationAppManager, COURSE_NOTIFICATION_TYPES, \ - NotificationTypeManager, filter_notification_types_by_app +from .base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES, filter_notification_types_by_app from .events import ( notification_preference_update_event, notification_read_event, @@ -31,10 +30,8 @@ from .serializers import ( NotificationSerializer, UserNotificationPreferenceUpdateAllSerializer, - add_info_to_notification_config, add_non_editable_in_preference ) -from .tasks import create_notification_preference from .utils import ( get_show_notifications_tray, exclude_inaccessible_preferences, create_account_notification_pref_if_not_exists @@ -252,202 +249,6 @@ def preference_update_from_encrypted_username_view(request, username, patch=""): return Response({"result": "success"}, status=status.HTTP_200_OK) -@allow_any_authenticated_user() -class NotificationPreferencesView(APIView): - """ - API view to retrieve and structure the notification preferences for the - authenticated user. - """ - - def get(self, request): - """ - Handles GET requests to retrieve notification preferences. - - This method fetches the user's active notification preferences and - merges them with a default structure provided by NotificationAppManager. - This provides a complete view of all possible notifications and the - user's current settings for them. - - Returns: - Response: A DRF Response object containing the structured - notification preferences or an error message. - """ - user_preferences_qs = NotificationPreference.objects.filter(user=request.user) - user_preferences_map = {pref.type: pref for pref in user_preferences_qs} - - # Ensure all notification types are present in the user's preferences. - # If any are missing, create them with default values. - diff = set(COURSE_NOTIFICATION_TYPES.keys()) - set(user_preferences_map.keys()) - missing_types = [] - for missing_type in diff: - new_pref = create_notification_preference( - user_id=request.user.id, - notification_type=missing_type, - - ) - missing_types.append(new_pref) - user_preferences_map[missing_type] = new_pref - if missing_types: - NotificationPreference.objects.bulk_create(missing_types) - - # If no user preferences are found, return an error response. - if not user_preferences_map: - return Response({ - 'status': 'error', - 'message': 'No active notification preferences found for this user.' - }, status=status.HTTP_404_NOT_FOUND) - - # Get the structured preferences from the NotificationAppManager. - # This will include all apps and their notification types. - structured_preferences = NotificationAppManager().get_notification_app_preferences() - - for app_name, app_settings in structured_preferences.items(): - notification_types = app_settings.get('notification_types', {}) - - # Process all notification types (core and non-core) in a single loop. - for type_name, type_details in notification_types.items(): - if type_name == 'core': - if structured_preferences[app_name]['core_notification_types']: - # If the app has core notification types, use the first one as the type name. - # This assumes that the first core notification type is representative of the core settings. - notification_type = structured_preferences[app_name]['core_notification_types'][0] - else: - notification_type = 'core' - user_pref = user_preferences_map.get(notification_type) - else: - user_pref = user_preferences_map.get(type_name) - if user_pref: - # If a preference exists, update the dictionary for this type. - # This directly modifies the 'type_details' dictionary. - type_details['web'] = user_pref.web - type_details['email'] = user_pref.email - type_details['push'] = user_pref.push - type_details['email_cadence'] = user_pref.email_cadence - exclude_inaccessible_preferences(structured_preferences, request.user) - structured_preferences = add_non_editable_in_preference( - add_info_to_notification_config(structured_preferences) - ) - return Response({ - 'status': 'success', - 'message': 'Notification preferences retrieved successfully.', - 'show_preferences': get_show_notifications_tray(), - 'data': structured_preferences - }, status=status.HTTP_200_OK) - - def put(self, request): - """ - Handles PUT requests to update notification preferences. - - This method updates the user's notification preferences based on the - provided data in the request body. It expects a dictionary with - notification types and their settings. - - Returns: - Response: A DRF Response object indicating success or failure. - """ - # Validate incoming data - serializer = UserNotificationPreferenceUpdateAllSerializer(data=request.data) - if not serializer.is_valid(): - return Response({ - 'status': 'error', - 'message': serializer.errors - }, status=status.HTTP_400_BAD_REQUEST) - - # Get validated data for easier access - validated_data = serializer.validated_data - - # Build query set based on notification type - query_set = NotificationPreference.objects.filter(user_id=request.user.id) - - if validated_data['notification_type'] == 'core': - # Get core notification types for the app - __, core_types = NotificationTypeManager().get_notification_app_preference( - notification_app=validated_data['notification_app'] - ) - query_set = query_set.filter(type__in=core_types) - else: - # Filter by single notification type - query_set = query_set.filter(type=validated_data['notification_type']) - - # Prepare update data based on channel type - updated_data = self._prepare_update_data(validated_data) - - # Update preferences - query_set.update(**updated_data) - - # Log the event - self._log_preference_update_event(request.user, validated_data) - - # Prepare and return response - response_data = self._prepare_response_data(validated_data) - return Response(response_data, status=status.HTTP_200_OK) - - def _prepare_update_data(self, validated_data): - """ - Prepare the data dictionary for updating notification preferences. - - Args: - validated_data (dict): Validated serializer data - - Returns: - dict: Dictionary with update data - """ - channel = validated_data['notification_channel'] - - if channel == 'email_cadence': - return {channel: validated_data['email_cadence']} - else: - return {channel: validated_data['value']} - - def _log_preference_update_event(self, user, validated_data): - """ - Log the notification preference update event. - - Args: - user: The user making the update - validated_data (dict): Validated serializer data - """ - event_data = { - 'notification_app': validated_data['notification_app'], - 'notification_type': validated_data['notification_type'], - 'notification_channel': validated_data['notification_channel'], - 'value': validated_data.get('value'), - 'email_cadence': validated_data.get('email_cadence'), - } - notification_preference_update_event(user, [], event_data) - - def _prepare_response_data(self, validated_data): - """ - Prepare the response data dictionary. - - Args: - validated_data (dict): Validated serializer data - - Returns: - dict: Response data dictionary - """ - email_cadence = validated_data.get('email_cadence', None) - # Determine the updated value - updated_value = validated_data.get('value', email_cadence if email_cadence else None) - - # Determine the channel - channel = validated_data.get('notification_channel') - if not channel and validated_data.get('email_cadence'): - channel = 'email_cadence' - - return { - 'status': 'success', - 'message': 'Notification preferences update completed', - 'show_preferences': get_show_notifications_tray(), - 'data': { - 'updated_value': updated_value, - 'notification_type': validated_data['notification_type'], - 'channel': channel, - 'app': validated_data['notification_app'], - } - } - - @allow_any_authenticated_user() class NotificationPreferencesViewV3(APIView): """ @@ -460,7 +261,7 @@ def get(self, request): Handles GET requests to retrieve notification preferences. This method fetches the user's active notification preferences and - merges them with a default structure provided by NotificationAppManager. + merges them with a default structure provided. This provides a complete view of all possible notifications and the user's current settings for them. @@ -532,7 +333,6 @@ def put(self, request): query_set = NotificationPreference.objects.filter(user_id=request.user.id) if validated_data['notification_type'] == 'grouped_notification': - # Get core notification types for the app grouped_types = filter_notification_types_by_app(validated_data['notification_app'], use_app_defaults=True) query_set = query_set.filter(type__in=grouped_types.keys()) else: