Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: added api to update all notification preferences for user (#35795)

Browse Source
This commit is contained in:
Ahtisham Shahid
2024-12-19 19:04:16 +05:00
committed by GitHub
parent 930989e5e6
commit 85a5890dd1
7 changed files with 1001 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
openedx/core/djangoapps/notifications/email/utils.py
View File

@@ -9,7 +9,7 @@ from django.conf import settings
from django.contrib.auth import get_user_model
from django.shortcuts import get_object_or_404
from pytz import utc
from waffle import get_waffle_flag_model # pylint: disable=invalid-django-waffle-import
from waffle import get_waffle_flag_model # pylint: disable=invalid-django-waffle-import
from common.djangoapps.student.models import CourseEnrollment
from lms.djangoapps.branding.api import get_logo_url_for_email
@@ -29,7 +29,6 @@ from xmodule.modulestore.django import modulestore
from .notification_icons import NotificationTypeIcons
User = get_user_model()
@@ -370,14 +369,6 @@ def update_user_preferences_from_patch(encrypted_username, encrypted_patch):
"""
return True if param_name is None else name == param_name
def is_editable(app_name, notification_type, channel):
"""
Returns if notification type channel is editable
"""
if notification_type == 'core':
return channel not in COURSE_NOTIFICATION_APPS[app_name]['non_editable']
return channel not in COURSE_NOTIFICATION_TYPES[notification_type]['non_editable']
def get_default_cadence_value(app_name, notification_type):
"""
Returns default email cadence value
@@ -417,9 +408,18 @@ def update_user_preferences_from_patch(encrypted_username, encrypted_patch):
for channel in ['web', 'email', 'push']:
if not is_name_match(channel, channel_value):
continue
if is_editable(app_name, noti_type, channel):
if is_notification_type_channel_editable(app_name, noti_type, channel):
type_prefs[channel] = pref_value
if channel == 'email' and pref_value and type_prefs.get('email_cadence') == EmailCadence.NEVER:
type_prefs['email_cadence'] = get_default_cadence_value(app_name, noti_type)
preference.save()
notification_preference_unsubscribe_event(user)
def is_notification_type_channel_editable(app_name, notification_type, channel):
"""
Returns if notification type channel is editable
"""
if notification_type == 'core':
return channel not in COURSE_NOTIFICATION_APPS[app_name]['non_editable']
return channel not in COURSE_NOTIFICATION_TYPES[notification_type]['non_editable']

116
openedx/core/djangoapps/notifications/serializers.py
View File

@@ -1,6 +1,7 @@
"""
Serializers for the notifications API.
"""
from django.core.exceptions import ValidationError
from rest_framework import serializers
@@ -9,9 +10,12 @@ from openedx.core.djangoapps.content.course_overviews.models import CourseOvervi
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
Notification,
get_notification_channels, get_additional_notification_channel_settings
get_additional_notification_channel_settings,
get_notification_channels
)
from .base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES, EmailCadence
from .email.utils import is_notification_type_channel_editable
from .utils import remove_preferences_with_no_access
@@ -202,3 +206,113 @@ class NotificationSerializer(serializers.ModelSerializer):
'last_seen',
'created',
)
def validate_email_cadence(email_cadence: str) -> str:
"""
Validate email cadence value.
"""
if EmailCadence.get_email_cadence_value(email_cadence) is None:
raise ValidationError(f'{email_cadence} is not a valid email cadence.')
return email_cadence
def validate_notification_app(notification_app: str) -> str:
"""
Validate notification app value.
"""
if not COURSE_NOTIFICATION_APPS.get(notification_app):
raise ValidationError(f'{notification_app} is not a valid notification app.')
return notification_app
def validate_notification_app_enabled(notification_app: str) -> str:
"""
Validate notification app is enabled.
"""
if COURSE_NOTIFICATION_APPS.get(notification_app) and COURSE_NOTIFICATION_APPS.get(notification_app)['enabled']:
return notification_app
raise ValidationError(f'{notification_app} is not a valid notification app.')
def validate_notification_type(notification_type: str) -> str:
"""
Validate notification type value.
"""
if not COURSE_NOTIFICATION_TYPES.get(notification_type):
raise ValidationError(f'{notification_type} is not a valid notification type.')
return notification_type
def validate_notification_channel(notification_channel: str) -> str:
"""
Validate notification channel value.
"""
valid_channels = set(get_notification_channels()) | set(get_additional_notification_channel_settings())
if notification_channel not in valid_channels:
raise ValidationError(f'{notification_channel} is not a valid notification channel setting.')
return notification_channel
class UserNotificationPreferenceUpdateAllSerializer(serializers.Serializer):
"""
Serializer for user notification preferences update with custom field validators.
"""
notification_app = serializers.CharField(
required=True,
validators=[validate_notification_app, validate_notification_app_enabled]
)
value = serializers.BooleanField(required=False)
notification_type = serializers.CharField(
required=True,
)
notification_channel = serializers.CharField(
required=False,
validators=[validate_notification_channel]
)
email_cadence = serializers.CharField(
required=False,
validators=[validate_email_cadence]
)
def validate(self, attrs):
"""
Cross-field validation for notification preference update.
"""
notification_app = attrs.get('notification_app')
notification_type = attrs.get('notification_type')
notification_channel = attrs.get('notification_channel')
email_cadence = attrs.get('email_cadence')
# Validate email_cadence requirements
if email_cadence and not notification_type:
raise ValidationError({
'notification_type': 'notification_type is required for email_cadence.'
})
# Validate notification_channel requirements
if not email_cadence and notification_type and not notification_channel:
raise ValidationError({
'notification_channel': 'notification_channel is required for notification_type.'
})
# Validate notification type
if all([not COURSE_NOTIFICATION_TYPES.get(notification_type), notification_type != "core"]):
raise ValidationError(f'{notification_type} is not a valid notification type.')
# Validate notification type and channel is editable
if notification_channel and notification_type:
if not is_notification_type_channel_editable(
notification_app,
notification_type,
notification_channel
):
raise ValidationError({
'notification_channel': (
f'{notification_channel} is not editable for notification type '
f'{notification_type}.'
)
})
return attrs

288
openedx/core/djangoapps/notifications/tests/test_utils.py Normal file
View File

@@ -0,0 +1,288 @@
"""
Test cases for the notification utility functions.
"""
import unittest
from openedx.core.djangoapps.notifications.utils import aggregate_notification_configs
class TestAggregateNotificationConfigs(unittest.TestCase):
"""
Test cases for the aggregate_notification_configs function.
"""
def test_empty_configs_list_returns_default(self):
"""
If the configs list is empty, the default config should be returned.
"""
default_config = [{
"grading": {
"enabled": False,
"non_editable": {},
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Daily"
}
}
}
}]
result = aggregate_notification_configs(default_config)
assert result == default_config[0]
def test_enable_notification_type(self):
"""
If a config enables a notification type, it should be enabled in the result.
"""
config_list = [
{
"grading": {
"enabled": False,
"non_editable": {},
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Weekly"
}
}
}
},
{
"grading": {
"enabled": True,
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Weekly"
}
}
}
}]
result = aggregate_notification_configs(config_list)
assert result["grading"]["enabled"] is True
assert result["grading"]["notification_types"]["core"]["web"] is True
assert result["grading"]["notification_types"]["core"]["push"] is True
assert result["grading"]["notification_types"]["core"]["email"] is True
# Use default email_cadence
assert result["grading"]["notification_types"]["core"]["email_cadence"] == "Weekly"
def test_merge_core_notification_types(self):
"""
Core notification types should be merged across configs.
"""
config_list = [
{
"discussion": {
"enabled": True,
"core_notification_types": ["new_comment"],
"notification_types": {}
}
},
{
"discussion": {
"core_notification_types": ["new_response", "new_comment"]
}
}]
result = aggregate_notification_configs(config_list)
assert set(result["discussion"]["core_notification_types"]) == {
"new_comment", "new_response"
}
def test_multiple_configs_aggregate(self):
"""
Multiple configs should be aggregated together.
"""
config_list = [
{
"updates": {
"enabled": False,
"notification_types": {
"course_updates": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Weekly"
}
}
}
},
{
"updates": {
"enabled": True,
"notification_types": {
"course_updates": {
"web": True,
"email_cadence": "Weekly"
}
}
}
},
{
"updates": {
"notification_types": {
"course_updates": {
"push": True,
"email_cadence": "Weekly"
}
}
}
}
]
result = aggregate_notification_configs(config_list)
assert result["updates"]["enabled"] is True
assert result["updates"]["notification_types"]["course_updates"]["web"] is True
assert result["updates"]["notification_types"]["course_updates"]["push"] is True
assert result["updates"]["notification_types"]["course_updates"]["email"] is False
# Use default email_cadence
assert result["updates"]["notification_types"]["course_updates"]["email_cadence"] == "Weekly"
def test_ignore_unknown_notification_types(self):
"""
Unknown notification types should be ignored.
"""
config_list = [
{
"grading": {
"enabled": False,
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Daily"
}
}
}
},
{
"grading": {
"notification_types": {
"unknown_type": {
"web": True,
"push": True,
"email": True
}
}
}
}]
result = aggregate_notification_configs(config_list)
assert "unknown_type" not in result["grading"]["notification_types"]
assert result["grading"]["notification_types"]["core"]["web"] is False
def test_ignore_unknown_categories(self):
"""
Unknown categories should be ignored.
"""
config_list = [
{
"grading": {
"enabled": False,
"notification_types": {}
}
},
{
"unknown_category": {
"enabled": True,
"notification_types": {}
}
}]
result = aggregate_notification_configs(config_list)
assert "unknown_category" not in result
assert result["grading"]["enabled"] is False
def test_preserves_default_structure(self):
"""
The resulting config should have the same structure as the default config.
"""
config_list = [
{
"discussion": {
"enabled": False,
"non_editable": {"core": ["web"]},
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Weekly"
}
},
"core_notification_types": []
}
},
{
"discussion": {
"enabled": True,
"extra_field": "should_not_appear"
}
}
]
result = aggregate_notification_configs(config_list)
assert set(result["discussion"].keys()) == {
"enabled", "non_editable", "notification_types", "core_notification_types"
}
assert "extra_field" not in result["discussion"]
def test_if_email_cadence_has_diff_set_mix_as_value(self):
"""
If email_cadence is different in the configs, set it to "Mixed".
"""
config_list = [
{
"grading": {
"enabled": False,
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Daily"
}
}
}
},
{
"grading": {
"enabled": True,
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Weekly"
}
}
}
},
{
"grading": {
"notification_types": {
"core": {
"email_cadence": "Monthly"
}
}
}
}
]
result = aggregate_notification_configs(config_list)
assert result["grading"]["notification_types"]["core"]["email_cadence"] == "Mixed"

315
openedx/core/djangoapps/notifications/tests/test_views.py
View File

@@ -2,11 +2,14 @@
Tests for the views in the notifications app.
"""
import json
from copy import deepcopy
from datetime import datetime, timedelta
from unittest import mock
from unittest.mock import patch
import ddt
from django.conf import settings
from django.contrib.auth import get_user_model
from django.test.utils import override_settings
from django.urls import reverse
from edx_toggles.toggles.testutils import override_waffle_flag
@@ -27,19 +30,21 @@ from openedx.core.djangoapps.django_comment_common.models import (
FORUM_ROLE_MODERATOR
)
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx.core.djangoapps.notifications.email.utils import encrypt_object, encrypt_string
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
Notification,
get_course_notification_preference_config_version
)
from openedx.core.djangoapps.notifications.serializers import NotificationCourseEnrollmentSerializer
from openedx.core.djangoapps.notifications.email.utils import encrypt_object, encrypt_string
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
from ..base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES, NotificationAppManager
from ..utils import get_notification_types_with_visibility_settings
User = get_user_model()
@ddt.ddt
class CourseEnrollmentListViewTest(ModuleStoreTestCase):
@@ -903,6 +908,7 @@ class UpdatePreferenceFromEncryptedDataView(ModuleStoreTestCase):
"""
Tests if preference is updated when encrypted url is hit
"""
def setUp(self):
"""
Setup test case
@@ -968,3 +974,310 @@ def remove_notifications_with_visibility_settings(expected_response):
notification_type
)
return expected_response
class UpdateAllNotificationPreferencesViewTests(APITestCase):
"""
Tests for the UpdateAllNotificationPreferencesView.
"""
def setUp(self):
# Create test user
self.user = User.objects.create_user(
username='testuser',
password='testpass123'
)
self.client = APIClient()
self.client.force_authenticate(user=self.user)
self.url = reverse('update-all-notification-preferences')
# Complex notification config structure
self.base_config = {
"grading": {
"enabled": True,
"non_editable": {},
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
},
"ora_staff_notification": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Daily"
}
},
"core_notification_types": []
},
"updates": {
"enabled": True,
"non_editable": {},
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
},
"course_updates": {
"web": True,
"push": True,
"email": False,
"email_cadence": "Daily"
}
},
"core_notification_types": []
},
"discussion": {
"enabled": True,
"non_editable": {
"core": ["web"]
},
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
},
"content_reported": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
},
"new_question_post": {
"web": True,
"push": False,
"email": False,
"email_cadence": "Daily"
},
"new_discussion_post": {
"web": True,
"push": False,
"email": False,
"email_cadence": "Daily"
}
},
"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"
]
}
}
# Create test notification preferences
self.preferences = []
for i in range(3):
pref = CourseNotificationPreference.objects.create(
user=self.user,
course_id=f'course-v1:TestX+Test{i}+2024',
notification_preference_config=deepcopy(self.base_config),
is_active=True
)
self.preferences.append(pref)
# Create an inactive preference
self.inactive_pref = CourseNotificationPreference.objects.create(
user=self.user,
course_id='course-v1:TestX+Inactive+2024',
notification_preference_config=deepcopy(self.base_config),
is_active=False
)
def test_update_discussion_notification(self):
"""
Test updating discussion notification settings
"""
data = {
'notification_app': 'discussion',
'notification_type': 'content_reported',
'notification_channel': 'push',
'value': False
}
response = self.client.post(self.url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['status'], 'success')
self.assertEqual(response.data['data']['total_updated'], 3)
# Verify database updates
for pref in CourseNotificationPreference.objects.filter(is_active=True):
self.assertFalse(
pref.notification_preference_config['discussion']['notification_types']['content_reported']['push']
)
def test_update_non_editable_field(self):
"""
Test attempting to update a non-editable field
"""
data = {
'notification_app': 'discussion',
'notification_type': 'core',
'notification_channel': 'web',
'value': False
}
response = self.client.post(self.url, data, format='json')
# Should fail because 'web' is non-editable for 'core' in discussion
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data['status'], 'error')
# Verify database remains unchanged
for pref in CourseNotificationPreference.objects.filter(is_active=True):
self.assertTrue(
pref.notification_preference_config['discussion']['notification_types']['core']['web']
)
def test_update_email_cadence(self):
"""
Test updating email cadence setting
"""
data = {
'notification_app': 'discussion',
'notification_type': 'content_reported',
'email_cadence': 'Weekly'
}
response = self.client.post(self.url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['status'], 'success')
# Verify database updates
for pref in CourseNotificationPreference.objects.filter(is_active=True):
notification_type = pref.notification_preference_config['discussion']['notification_types'][
'content_reported']
self.assertEqual(
notification_type['email_cadence'],
'Weekly'
)
@patch.dict('openedx.core.djangoapps.notifications.serializers.COURSE_NOTIFICATION_APPS', {
**COURSE_NOTIFICATION_APPS,
'grading': {
'enabled': False,
'core_info': 'Notifications for submission grading.',
'core_web': True,
'core_email': True,
'core_push': True,
'core_email_cadence': 'Daily',
'non_editable': []
}
})
def test_update_disabled_app(self):
"""
Test updating notification for a disabled app
"""
# Disable the grading app in all preferences
for pref in self.preferences:
config = pref.notification_preference_config
config['grading']['enabled'] = False
pref.notification_preference_config = config
pref.save()
data = {
'notification_app': 'grading',
'notification_type': 'core',
'notification_channel': 'email',
'value': False
}
response = self.client.post(self.url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data['status'], 'error')
def test_invalid_serializer_data(self):
"""
Test handling of invalid input data
"""
test_cases = [
{
'notification_app': 'invalid_app',
'notification_type': 'core',
'notification_channel': 'push',
'value': False
},
{
'notification_app': 'discussion',
'notification_type': 'invalid_type',
'notification_channel': 'push',
'value': False
},
{
'notification_app': 'discussion',
'notification_type': 'core',
'notification_channel': 'invalid_channel',
'value': False
},
{
'notification_app': 'discussion',
'notification_type': 'core',
'notification_channel': 'email_cadence',
'value': 'Invalid_Cadence'
}
]
for test_case in test_cases:
response = self.client.post(self.url, test_case, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
class GetAggregateNotificationPreferencesTest(APITestCase):
"""
Tests for the GetAggregateNotificationPreferences API view.
"""
def setUp(self):
# Set up a user and API client
self.user = User.objects.create_user(username='testuser', password='testpass')
self.client = APIClient()
self.client.force_authenticate(user=self.user)
self.url = reverse('notification-preferences-aggregated') # Adjust with the actual name
def test_no_active_notification_preferences(self):
"""
Test case: No active notification preferences found for the user
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertEqual(response.data['status'], 'error')
self.assertEqual(response.data['message'], 'No active notification preferences found')
@patch('openedx.core.djangoapps.notifications.views.aggregate_notification_configs')
def test_with_active_notification_preferences(self, mock_aggregate):
"""
Test case: Active notification preferences found for the user
"""
# Mock aggregate_notification_configs for a controlled output
mock_aggregate.return_value = {'mocked': 'data'}
# Create active notification preferences for the user
CourseNotificationPreference.objects.create(
user=self.user,
is_active=True,
notification_preference_config={'example': 'config'}
)
response = self.client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['status'], 'success')
self.assertEqual(response.data['message'], 'Notification preferences retrieved')
self.assertEqual(response.data['data'], {'mocked': 'data'})
def test_unauthenticated_user(self):
"""
Test case: Request without authentication
"""
# Test case: Request without authentication
self.client.logout() # Remove authentication
response = self.client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

14
openedx/core/djangoapps/notifications/urls.py
View File

@@ -11,13 +11,13 @@ from .views import (
NotificationCountView,
NotificationListAPIView,
NotificationReadAPIView,
UpdateAllNotificationPreferencesView,
UserNotificationPreferenceView,
preference_update_from_encrypted_username_view,
preference_update_from_encrypted_username_view, AggregatedNotificationPreferences
)
router = routers.DefaultRouter()
urlpatterns = [
path('enrollments/', CourseEnrollmentListView.as_view(), name='enrollment-list'),
re_path(
@@ -25,6 +25,11 @@ urlpatterns = [
UserNotificationPreferenceView.as_view(),
name='notification-preferences'
),
path(
'configurations/',
AggregatedNotificationPreferences.as_view(),
name='notification-preferences-aggregated'
),
path('', NotificationListAPIView.as_view(), name='notifications-list'),
path('count/', NotificationCountView.as_view(), name='notifications-count'),
path(
@@ -35,6 +40,11 @@ urlpatterns = [
path('read/', NotificationReadAPIView.as_view(), name='notifications-read'),
path('preferences/update/<str:username>/<str:patch>/', preference_update_from_encrypted_username_view,
name='preference_update_from_encrypted_username_view'),
path(
'preferences/update-all/',
UpdateAllNotificationPreferencesView.as_view(),
name='update-all-notification-preferences'
),
]
urlpatterns += router.urls

115
openedx/core/djangoapps/notifications/utils.py
View File

@@ -1,11 +1,12 @@
"""
Utils function for notifications app
"""
from typing import Dict, List
import copy
from typing import Dict, List, Set
from common.djangoapps.student.models import CourseAccessRole, CourseEnrollment
from openedx.core.djangoapps.django_comment_common.models import Role
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS, ENABLE_NEW_NOTIFICATION_VIEW
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NEW_NOTIFICATION_VIEW, ENABLE_NOTIFICATIONS
from openedx.core.lib.cache_utils import request_cached
@@ -158,3 +159,113 @@ def clean_arguments(kwargs):
if kwargs.get('created', {}):
clean_kwargs.update(kwargs.get('created'))
return clean_kwargs
def update_notification_types(
app_config: Dict,
user_app_config: Dict,
) -> None:
"""
Update notification types for a specific category configuration.
"""
if "notification_types" not in user_app_config:
return
for type_key, type_config in user_app_config["notification_types"].items():
if type_key not in app_config["notification_types"]:
continue
update_notification_fields(
app_config["notification_types"][type_key],
type_config,
)
def update_notification_fields(
target_config: Dict,
source_config: Dict,
) -> None:
"""
Update individual notification fields (web, push, email) and email_cadence.
"""
for field in ["web", "push", "email"]:
if field in source_config:
target_config[field] |= source_config[field]
if "email_cadence" in source_config:
if isinstance(target_config["email_cadence"], str) or not target_config["email_cadence"]:
target_config["email_cadence"] = set()
target_config["email_cadence"].add(source_config["email_cadence"])
def update_core_notification_types(app_config: Dict, user_config: Dict) -> None:
"""
Update core notification types by merging existing and new types.
"""
if "core_notification_types" not in user_config:
return
existing_types: Set = set(app_config.get("core_notification_types", []))
existing_types.update(user_config["core_notification_types"])
app_config["core_notification_types"] = list(existing_types)
def process_app_config(
app_config: Dict,
user_config: Dict,
app: str,
default_config: Dict,
) -> None:
"""
Process a single category configuration against another config.
"""
if app not in user_config:
return
user_app_config = user_config[app]
# Update enabled status
app_config["enabled"] |= user_app_config.get("enabled", False)
# Update core notification types
update_core_notification_types(app_config, user_app_config)
# Update notification types
update_notification_types(app_config, user_app_config)
def aggregate_notification_configs(existing_user_configs: List[Dict]) -> Dict:
"""
Update default notification config with values from other configs.
Rules:
1. Start with default config as base
2. If any value is True in other configs, make it True
3. Set email_cadence to "Mixed" if different cadences found, else use default
Args:
existing_user_configs: List of notification config dictionaries to apply
Returns:
Updated config following the same structure
"""
if not existing_user_configs:
return {}
result_config = copy.deepcopy(existing_user_configs[0])
apps = result_config.keys()
for app in apps:
app_config = result_config[app]
for user_config in existing_user_configs:
process_app_config(app_config, user_config, app, existing_user_configs[0])
# if email_cadence is mixed, set it to "Mixed"
for app in result_config:
for type_key, type_config in result_config[app]["notification_types"].items():
if len(type_config["email_cadence"]) > 1:
result_config[app]["notification_types"][type_key]["email_cadence"] = "Mixed"
else:
result_config[app]["notification_types"][type_key]["email_cadence"] = (
result_config[app]["notification_types"][type_key]["email_cadence"].pop())
return result_config

155
openedx/core/djangoapps/notifications/views.py
View File

@@ -1,9 +1,11 @@
"""
Views for the notifications API.
"""
import copy
from datetime import datetime, timedelta
from django.conf import settings
from django.db import transaction
from django.db.models import Count
from django.shortcuts import get_object_or_404
from django.utils.translation import gettext as _
@@ -17,10 +19,7 @@ from rest_framework.views import APIView
from common.djangoapps.student.models import CourseEnrollment
from openedx.core.djangoapps.notifications.email.utils import update_user_preferences_from_patch
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
get_course_notification_preference_config_version
)
from openedx.core.djangoapps.notifications.models import get_course_notification_preference_config_version
from openedx.core.djangoapps.notifications.permissions import allow_any_authenticated_user
from .base_notification import COURSE_NOTIFICATION_APPS
@@ -32,14 +31,15 @@ from .events import (
notification_tray_opened_event,
notifications_app_all_read_event
)
from .models import Notification
from .models import CourseNotificationPreference, Notification
from .serializers import (
NotificationCourseEnrollmentSerializer,
NotificationSerializer,
UserCourseNotificationPreferenceSerializer,
UserNotificationPreferenceUpdateSerializer,
UserNotificationPreferenceUpdateAllSerializer,
UserNotificationPreferenceUpdateSerializer
)
from .utils import get_show_notifications_tray, get_is_new_notification_view_enabled
from .utils import get_is_new_notification_view_enabled, get_show_notifications_tray, aggregate_notification_configs
@allow_any_authenticated_user()
@@ -444,3 +444,144 @@ def preference_update_from_encrypted_username_view(request, username, patch):
"""
update_user_preferences_from_patch(username, patch)
return Response({"result": "success"}, status=status.HTTP_200_OK)
@allow_any_authenticated_user()
class UpdateAllNotificationPreferencesView(APIView):
"""
API view for updating all notification preferences for the current user.
"""
def post(self, request):
"""
Update all notification preferences for the current user.
"""
# check if request have required params
serializer = UserNotificationPreferenceUpdateAllSerializer(data=request.data)
if not serializer.is_valid():
return Response({
'status': 'error',
'message': serializer.errors
}, status=status.HTTP_400_BAD_REQUEST)
# check if required config is not editable
try:
with transaction.atomic():
# Get all active notification preferences for the current user
notification_preferences = (
CourseNotificationPreference.objects
.select_for_update()
.filter(
user=request.user,
is_active=True
)
)
if not notification_preferences.exists():
return Response({
'status': 'error',
'message': 'No active notification preferences found'
}, status=status.HTTP_404_NOT_FOUND)
data = serializer.validated_data
app = data['notification_app']
email_cadence = data.get('email_cadence', None)
channel = data.get('notification_channel', 'email_cadence' if email_cadence else None)
notification_type = data['notification_type']
value = data.get('value', email_cadence if email_cadence else None)
updated_courses = []
errors = []
# Update each preference
for preference in notification_preferences:
try:
# Create a deep copy of the current config
updated_config = copy.deepcopy(preference.notification_preference_config)
# Check if the path exists and update the value
if (
updated_config.get(app, {})
.get('notification_types', {})
.get(notification_type, {})
.get(channel)
) is not None:
# Update the specific setting in the config
updated_config[app]['notification_types'][notification_type][channel] = value
# Update the notification preference
preference.notification_preference_config = updated_config
preference.save()
updated_courses.append({
'course_id': str(preference.course_id),
'current_setting': updated_config[app]['notification_types'][notification_type]
})
else:
errors.append({
'course_id': str(preference.course_id),
'error': f'Invalid path: {app}.notification_types.{notification_type}.{channel}'
})
except Exception as e:
errors.append({
'course_id': str(preference.course_id),
'error': str(e)
})
response_data = {
'status': 'success' if updated_courses else 'partial_success' if errors else 'error',
'message': 'Notification preferences update completed',
'data': {
'updated_value': value,
'notification_type': notification_type,
'channel': channel,
'app': app,
'successfully_updated_courses': updated_courses,
'total_updated': len(updated_courses),
'total_courses': notification_preferences.count()
}
}
if errors:
response_data['errors'] = errors
return Response(
response_data,
status=status.HTTP_200_OK if updated_courses else status.HTTP_400_BAD_REQUEST
)
except Exception as e:
return Response({
'status': 'error',
'message': str(e)
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
@allow_any_authenticated_user()
class AggregatedNotificationPreferences(APIView):
"""
API view for getting the aggregate notification preferences for the current user.
"""
def get(self, request):
"""
API view for getting the aggregate notification preferences for the current user.
"""
notification_preferences = CourseNotificationPreference.objects.filter(user=request.user, is_active=True)
if not notification_preferences.exists():
return Response({
'status': 'error',
'message': 'No active notification preferences found'
}, status=status.HTTP_404_NOT_FOUND)
notification_configs = notification_preferences.values_list('notification_preference_config', flat=True)
notification_configs = aggregate_notification_configs(
notification_configs
)
return Response({
'status': 'success',
'message': 'Notification preferences retrieved',
'data': notification_configs
}, status=status.HTTP_200_OK)