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

Ahtisham/remove core type from notifications (#37868)

Browse Source
This commit is contained in:
Ahtisham Shahid
2026-01-15 16:37:50 +05:00
committed by GitHub
parent 328b3ee3fa
commit cbdb721db0
13 changed files with 802 additions and 442 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
openedx/core/djangoapps/notifications/base_notification.py
View File

@@ -26,9 +26,9 @@ class NotificationType(TypedDict):
# Unique identifier for this notification type.
name: str
# Mark this as a core notification.
# When True, user preferences are taken from the notification app's `core_*` configuration,
# 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.
is_core: bool
use_app_defaults: bool
# Template string for notification content (see ./docs/templates.md).
# Wrap in gettext_lazy (_) for translation support.
content_template: str
@@ -36,8 +36,6 @@ class NotificationType(TypedDict):
# The values for these variables are passed to the templates when generating the notification.
# NOTE: this field is for documentation purposes only; it is not used.
content_context: dict[str, Any]
# Template used when delivering notifications via email.
email_template: str
filters: list[str]
# All fields below are required unless `is_core` is True.
@@ -67,20 +65,20 @@ _COURSE_NOTIFICATION_TYPES = {
'new_comment_on_response': {
'notification_app': 'discussion',
'name': 'new_comment_on_response',
'is_core': True,
'use_app_defaults': True,
'content_template': _('<{p}><{strong}>{replier_name}</{strong}> commented on your response to the post '
'<{strong}>{post_title}</{strong}></{p}>'),
'content_context': {
'post_title': 'Post title',
'replier_name': 'replier name',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
'new_comment': {
'notification_app': 'discussion',
'name': 'new_comment',
'is_core': True,
'use_app_defaults': True,
'content_template': _('<{p}><{strong}>{replier_name}</{strong}> commented on <{strong}>{author_name}'
'</{strong}> response to your post <{strong}>{post_title}</{strong}></{p}>'),
'content_context': {
@@ -88,13 +86,13 @@ _COURSE_NOTIFICATION_TYPES = {
'author_name': 'author name',
'replier_name': 'replier name',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
'new_response': {
'notification_app': 'discussion',
'name': 'new_response',
'is_core': True,
'use_app_defaults': True,
'content_template': _('<{p}><{strong}>{replier_name}</{strong}> responded to your '
'post <{strong}>{post_title}</{strong}></{p}>'),
'grouped_content_template': _('<{p}><{strong}>{replier_name}</{strong}> and others have responded to your post '
@@ -103,13 +101,13 @@ _COURSE_NOTIFICATION_TYPES = {
'post_title': 'Post title',
'replier_name': 'replier name',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
'new_discussion_post': {
'notification_app': 'discussion',
'name': 'new_discussion_post',
'is_core': False,
'info': '',
'web': False,
'email': False,
@@ -123,13 +121,13 @@ _COURSE_NOTIFICATION_TYPES = {
'post_title': 'Post title',
'username': 'Post author name',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
'new_question_post': {
'notification_app': 'discussion',
'name': 'new_question_post',
'is_core': False,
'info': '',
'web': False,
'email': False,
@@ -141,15 +139,13 @@ _COURSE_NOTIFICATION_TYPES = {
'post_title': 'Post title',
'username': 'Post author name',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
'response_on_followed_post': {
'notification_app': 'discussion',
'name': 'response_on_followed_post',
'is_core': True,
'info': '',
'non_editable': [],
'use_app_defaults': True,
'content_template': _('<{p}><{strong}>{replier_name}</{strong}> responded to a post you’re following: '
'<{strong}>{post_title}</{strong}></{p}>'),
'grouped_content_template': _('<{p}><{strong}>{replier_name}</{strong}> and others responded to a post you’re '
@@ -158,15 +154,13 @@ _COURSE_NOTIFICATION_TYPES = {
'post_title': 'Post title',
'replier_name': 'replier name',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
'comment_on_followed_post': {
'notification_app': 'discussion',
'name': 'comment_on_followed_post',
'is_core': True,
'info': '',
'non_editable': [],
'use_app_defaults': True,
'content_template': _('<{p}><{strong}>{replier_name}</{strong}> commented on <{strong}>{author_name}'
'</{strong}> response in a post you’re following <{strong}>{post_title}'
'</{strong}></{p}>'),
@@ -175,13 +169,13 @@ _COURSE_NOTIFICATION_TYPES = {
'author_name': 'author name',
'replier_name': 'replier name',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
'content_reported': {
'notification_app': 'discussion',
'name': 'content_reported',
'is_core': False,
'info': '',
'web': True,
'email': True,
@@ -196,42 +190,38 @@ _COURSE_NOTIFICATION_TYPES = {
'author_name': 'author name',
'replier_name': 'replier name',
},
'email_template': '',
'visible_to': [FORUM_ROLE_ADMINISTRATOR, FORUM_ROLE_MODERATOR, FORUM_ROLE_COMMUNITY_TA]
},
'response_endorsed_on_thread': {
'notification_app': 'discussion',
'name': 'response_endorsed_on_thread',
'is_core': True,
'info': '',
'non_editable': [],
'use_app_defaults': True,
'content_template': _('<{p}><{strong}>{replier_name}\'s</{strong}> response has been endorsed in your post '
'<{strong}>{post_title}</{strong}></{p}>'),
'content_context': {
'post_title': 'Post title',
'replier_name': 'replier name',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
'response_endorsed': {
'notification_app': 'discussion',
'name': 'response_endorsed',
'is_core': True,
'info': '',
'non_editable': [],
'use_app_defaults': True,
'content_template': _('<{p}>Your response has been endorsed on the post <{strong}>{post_title}</{strong}></{'
'p}>'),
'content_context': {
'post_title': 'Post title',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
'course_updates': {
'notification_app': 'updates',
'name': 'course_updates',
'is_core': False,
'info': '',
'web': True,
'email': True,
@@ -242,13 +232,13 @@ _COURSE_NOTIFICATION_TYPES = {
'content_context': {
'course_update_content': 'Course update',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
'ora_staff_notifications': {
'notification_app': 'grading',
'name': 'ora_staff_notifications',
'is_core': False,
'info': 'Notifications for when a submission is made for ORA that includes staff grading step.',
'web': True,
'email': False,
@@ -262,14 +252,14 @@ _COURSE_NOTIFICATION_TYPES = {
'content_context': {
'ora_name': 'Name of ORA in course',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE],
'visible_to': [CourseStaffRole.ROLE, CourseInstructorRole.ROLE]
},
'ora_grade_assigned': {
'notification_app': 'grading',
'name': 'ora_grade_assigned',
'is_core': False,
'info': '',
'web': True,
'email': True,
@@ -283,13 +273,13 @@ _COURSE_NOTIFICATION_TYPES = {
'points_earned': 'Points earned',
'points_possible': 'Points possible',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE],
},
'new_instructor_all_learners_post': {
'notification_app': 'discussion',
'name': 'new_instructor_all_learners_post',
'is_core': False,
'info': '',
'web': True,
'email': True,
@@ -301,7 +291,7 @@ _COURSE_NOTIFICATION_TYPES = {
'content_context': {
'post_title': 'Post title',
},
'email_template': '',
'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
},
}
@@ -322,21 +312,21 @@ class NotificationApp(TypedDict):
"""
# Set to True to enable this app and linked notification types.
enabled: bool
# Description to be displayed about core notifications for this app.
# Description to be displayed about grouped notifications for this app.
# This string should be wrapped in the gettext_lazy function (imported as `_`) to support translation.
core_info: str
# Set to True to enable delivery for associated core notifications on web.
core_web: bool
# Set to True to enable delivery for associated core notifications via emails.
core_email: bool
# Set to True to enable delivery for associated core notifications via push notifications.
info: str
# Set to True to enable delivery for associated grouped notifications on web.
web: bool
# Set to True to enable delivery for associated grouped notifications via emails.
email: bool
# Set to True to enable delivery for associated grouped notifications via push notifications.
# NOTE: push notifications are not implemented yet
core_push: bool
# How often email notifications are sent for associated core notifications.
core_email_cadence: Literal[EmailCadence.DAILY, EmailCadence.WEEKLY, EmailCadence.IMMEDIATELY, EmailCadence.NEVER]
# Items in the list represent core notification delivery channels
push: bool
# How often email notifications are sent for associated grouped notifications.
email_cadence: Literal[EmailCadence.DAILY, EmailCadence.WEEKLY, EmailCadence.IMMEDIATELY, EmailCadence.NEVER]
# Items in the list represent grouped notification delivery channels
# where the user is blocked from changing from what is defined for the app here
# (see `core_web`, `core_email`, and `core_push` above).
# (see `web`, `email`, and `push` above).
non_editable: list[Literal["web", "email", "push"]]
@@ -344,30 +334,30 @@ class NotificationApp(TypedDict):
_COURSE_NOTIFICATION_APPS: dict[str, NotificationApp] = {
'discussion': {
'enabled': True,
'core_info': _('Notifications for responses and comments on your posts, and the ones you’re '
'following, including endorsements to your responses and on your posts.'),
'core_web': True,
'core_email': True,
'core_push': True,
'core_email_cadence': EmailCadence.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.'),
'web': True,
'email': True,
'push': True,
'email_cadence': EmailCadence.DAILY,
'non_editable': []
},
'updates': {
'enabled': True,
'core_info': _('Notifications for new announcements and updates from the course team.'),
'core_web': True,
'core_email': True,
'core_push': True,
'core_email_cadence': EmailCadence.DAILY,
'info': _('Notifications for new announcements and updates from the course team.'),
'web': True,
'email': True,
'push': True,
'email_cadence': EmailCadence.DAILY,
'non_editable': []
},
'grading': {
'enabled': True,
'core_info': _('Notifications for submission grading.'),
'core_web': True,
'core_email': True,
'core_push': True,
'core_email_cadence': EmailCadence.DAILY,
'info': _('Notifications for submission grading.'),
'web': True,
'email': True,
'push': True,
'email_cadence': EmailCadence.DAILY,
'non_editable': []
},
}
@@ -405,7 +395,7 @@ class NotificationTypeManager:
core_notification_types = []
non_core_notification_types = []
for notification_type in notification_types:
if notification_type.get('is_core', None):
if notification_type.get('use_app_defaults', None):
core_notification_types.append(notification_type)
else:
non_core_notification_types.append(notification_type)
@@ -450,10 +440,10 @@ class NotificationAppManager:
Adds core notification preference for the given notification app.
"""
notification_types['core'] = {
'web': notification_app_attrs.get('core_web', False),
'email': False if email_opt_out else notification_app_attrs.get('core_email', False),
'push': notification_app_attrs.get('core_push', False),
'email_cadence': notification_app_attrs.get('core_email_cadence', 'Daily'),
'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):
@@ -533,3 +523,38 @@ def get_default_values_of_preference(notification_app, notification_type):
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
"""
preferences = {}
for name, values in COURSE_NOTIFICATION_TYPES.items():
if values.get('use_app_defaults', None):
app_defaults = COURSE_NOTIFICATION_APPS[values['notification_app']]
preferences[name] = {**app_defaults, **values}
else:
preferences[name] = {**values}
return preferences
def filter_notification_types_by_app(app_name, use_app_defaults=None) -> dict[str, dict[str, Any]]:
"""
Filter notification types by app name and optionally by use_app_defaults flag.
Args:
app_name (str): The notification app name to filter by (e.g., 'discussion', 'grading', 'updates')
use_app_defaults (bool, optional): If provided, additionally filter by use_app_defaults value
Returns:
dict: Filtered dictionary containing only matching notification types
"""
notification_types = get_default_values_of_preferences()
if use_app_defaults is None:
return {k: v for k, v in notification_types.items()
if v.get('notification_app') == app_name}
return {k: v for k, v in notification_types.items()
if v.get('notification_app') == app_name
and v.get('use_app_defaults', False) == use_app_defaults}

68
openedx/core/djangoapps/notifications/docs/settings.md
View File

@@ -12,50 +12,47 @@ This document explains how to override notification preferences defaults for new
**Notification apps:** For organization purposes, notification preferences are grouped by apps/platform workflows that they are related to. As of now, we have 3 apps: discussions, grading, and updates. Note that there is no app-level on/off switch for notifications types in it.
**Bundled preferences:** Each app discussed above has a set of preferences that start with `core_*` (e.g., `core_web`, `core_email`, `core_email_cadence`). A notification type marked `is_core: True` inherits its preferences from the app's `core_*` preferences instead of using its own default preferences. This enables us to bundle several notification types under one row of preferences visible to the user. For example, in case of discussions, 7 notifications types are controlled by `core_*` preference of the discussions app, which appears on the Account Settings page as a single row labeled "Activity Notifications" (see details here: https://openedx.atlassian.net/wiki/spaces/OEPM/pages/4750475268/Current+state+of+OpenedX+Notifications#Activities%2C-Preferences-and-Defaults). This reduces clutter on the settings page.
**App-level defaults / Bundled preferences:** Each app discussed above has a set of app-level defaults (bundled preferences). A notification type marked `use_app_defaults: True` inherits its preferences from the app's app-level defaults instead of using its own default preferences. This enables grouping several notification types under one row of preferences visible to the user. For example, in case of discussions, 7 notification types are controlled by the app-level defaults of the discussions app, which appears on the Account Settings page as a single row labeled "Activity Notifications". This reduces clutter on the settings page.
### What you can override
The table below lists all the preferences that you can override for each notification along with possible values. You can override defaults for both using the following dictionaries in your `lms.yml`, `cms.yml`, or `settings.py`:
- Use `NOTIFICATION_TYPES_OVERRIDE` to override defaults for notifications.
- Use `NOTIFICATION_APPS_OVERRIDE` to override defaults for notifications marked as core (`is_core: True`). Instead of notification key, use the app key here. At present, there are 3 apps whose keys are as follows:
- discussion
- updates
- grading
- Use `NOTIFICATION_TYPES_OVERRIDE` to override defaults for individual notification types.
- Use `NOTIFICATION_APPS_OVERRIDE` to override app-level defaults for notification types marked as `use_app_defaults: True`. Use the internal keys listed below (`web`, `email`, `push`, `email_cadence`, `non_editable`).
| Key | Key for notifications marked as core | Type | Possible values | What it does |
|-----|--------------------------------------|------|-----------------|--------------|
| web | core_web | bool | True OR False | Determines if the user gets notifications in the tray. |
| email | core_email | bool | True OR False | Determines if the user gets notification in email. |
| push | core_push | bool | True OR False | Determines if user gets push notification in mobile apps (not implemented) |
| email_cadence | core_email_cadence | string | 'Immediately' OR 'Daily' OR 'Weekly' | Determines when a user receives email notification. |
| non_editable | non_editable | list of strings | Any subset of ['web','email','push'] e.g. ['email', 'web'] | Determines toggles of which of the 3 channels will not be editable by the user. |
| Key | Type | Possible values | What it does |
|-----|------|-----------------|--------------|
| web | bool | True OR False | Determines if the user gets notifications in the tray. |
| email | bool | True OR False | Determines if the user gets notification in email. |
| push | bool | True OR False | Determines if user gets push notification in mobile apps (not implemented) |
| email_cadence | string | 'Immediately' OR 'Daily' OR 'Weekly' OR 'Never' | Determines when a user receives email notification. |
| non_editable | list of strings | Any subset of ['web','email','push'] e.g. ['email', 'web'] | Determines toggles of which of the 3 channels will not be editable by the user. |
Notification keys are listed in the table below. More notifications may be added in the future. You can find notification keys in this code: https://github.com/openedx/edx-platform/blob/2aeac459945e3e11c153fdb5203ea020514548d5/openedx/core/djangoapps/notifications/base_notification.py#L66
| # | Notification app | Notification key | is_core True? | Appears on Account Settings page as |
|---|------------------|------------------|---------------|-------------------------------------|
| 1 | discussion | new_response | Yes | Activity Notifications |
| 2 | discussion | new_comment | Yes | Activity Notifications |
| 3 | discussion | new_comment_on_response | Yes | Activity Notifications |
| 4 | discussion | response_on_followed_post | Yes | Activity Notifications |
| 5 | discussion | comment_on_followed_post | Yes | Activity Notifications |
| 6 | discussion | response_endorsed_on_thread | Yes | Activity Notifications |
| 7 | discussion | response_endorsed | Yes | Activity Notifications |
| 8 | discussion | content_reported | No | Reported content |
| 9 | discussion | new_question_post | No | New question posts |
| 10 | discussion | new_discussion_post | No | New discussion posts |
| 11 | discussion | new_instructor_all_learners_post | No | New posts from instructors |
| 12 | updates | course_updates | No | Course updates |
| 13 | grading | ora_staff_notifications | No | New ORA submission for staff |
| 14 | grading | ora_grade_assigned | No | ORA grade received |
| # | Notification app | Notification key | uses app defaults | Appears on Account Settings page as |
|---|------------------|------------------|-------------------|-------------------------------------|
| 1 | discussion | new_response | True | Activity Notifications |
| 2 | discussion | new_comment | True | Activity Notifications |
| 3 | discussion | new_comment_on_response | True | Activity Notifications |
| 4 | discussion | response_on_followed_post | True | Activity Notifications |
| 5 | discussion | comment_on_followed_post | True | Activity Notifications |
| 6 | discussion | response_endorsed_on_thread | True | Activity Notifications |
| 7 | discussion | response_endorsed | True | Activity Notifications |
| 8 | discussion | content_reported | False | Reported content |
| 9 | discussion | new_question_post | False | New question posts |
| 10 | discussion | new_discussion_post | False | New discussion posts |
| 11 | discussion | new_instructor_all_learners_post | False | New posts from instructors |
| 12 | updates | course_updates | False | Course updates |
| 13 | grading | ora_staff_notifications | False | New ORA submission for staff |
| 14 | grading | ora_grade_assigned | False | ORA grade received |
### Example configuration for overriding notification preferences:
```python
NOTIFICATION_TYPES_OVERRIDE = {
# Turn off tray and and turn on email notifications for new discussion posts and set daily cadence.
# Turn off tray and turn on email notifications for new discussion posts and set daily cadence.
'new_discussion_post': {
'email': True,
'web': False,
@@ -69,14 +66,15 @@ NOTIFICATION_TYPES_OVERRIDE = {
}
```
### Example configuration for overriding notification preference marked as core:
### Example configuration for overriding app-level defaults for notifications marked as use_app_defaults: True
```python
NOTIFICATION_APPS_OVERRIDE = {
# Turn on tray and turn off email for 7 discussion notification types that appear on Account Settings page as "Activity Notifications".
# For the 'discussion' app, set tray on and email off for all app-default notifications.
'discussion': {
'core_email': False,
'core_web': True
'email': False,
'web': True,
'email_cadence': 'Immediately',
}
}
```
@@ -84,4 +82,4 @@ NOTIFICATION_APPS_OVERRIDE = {
### Why isn't my override working?
- See if you are using the exact key name (e.g. `new_discussion_post` and not `New_discussion_post`).
- If a notification is marked as core (`is_core: True`) in the code, it will ignore overrides in `NOTIFICATION_TYPES_OVERRIDE`. You must override using `NOTIFICATION_APPS_OVERRIDE` instead.
- If a notification is marked as `use_app_defaults: True` in the code, it will ignore overrides in `NOTIFICATION_TYPES_OVERRIDE`. You must override using `NOTIFICATION_APPS_OVERRIDE` instead.

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

@@ -15,7 +15,10 @@ from waffle import get_waffle_flag_model # pylint: disable=invalid-django-waffl
from lms.djangoapps.branding.api import get_logo_url_for_email
from lms.djangoapps.discussion.notification_prefs.views import UsernameCipher, UsernameDecryptionException
from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY
from openedx.core.djangoapps.notifications.base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES
from openedx.core.djangoapps.notifications.base_notification import (
COURSE_NOTIFICATION_TYPES,
get_default_values_of_preferences
)
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_EMAIL_NOTIFICATIONS
from openedx.core.djangoapps.notifications.email import ONE_CLICK_EMAIL_UNSUB_KEY
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
@@ -26,7 +29,7 @@ from openedx.core.djangoapps.site_configuration import helpers as configuration_
from xmodule.modulestore.django import modulestore
from .notification_icons import NotificationTypeIcons
from ..utils import create_account_notification_pref_if_not_exists
User = get_user_model()
@@ -291,79 +294,15 @@ def get_unique_course_ids(notifications):
return course_ids
def get_enabled_notification_types_for_cadence(preferences, cadence_type=EmailCadence.DAILY):
"""
Returns a dictionary that returns notification_types with cadence_types for course_ids
"""
if cadence_type not in [EmailCadence.DAILY, EmailCadence.WEEKLY]:
raise ValueError('Invalid cadence_type')
course_types = {}
for preference in preferences:
key = preference.course_id
value = []
config = preference.notification_preference_config
for app_data in config.values():
for notification_type, type_dict in app_data['notification_types'].items():
if (type_dict['email_cadence'] == cadence_type) and type_dict['email']:
value.append(notification_type)
if 'core' in value:
value.remove('core')
value.extend(app_data['core_notification_types'])
course_types[key] = value
return course_types
def filter_notification_with_email_enabled_preferences(notifications, preferences, cadence_type=EmailCadence.DAILY):
"""
Filter notifications for types with email cadence preference enabled
"""
enabled_course_prefs = get_enabled_notification_types_for_cadence(preferences, cadence_type)
filtered_notifications = []
for notification in notifications:
if notification.notification_type in enabled_course_prefs[notification.course_id]:
filtered_notifications.append(notification)
filtered_notifications.sort(key=lambda elem: elem.created, reverse=True)
return filtered_notifications
def create_missing_account_level_preferences(notifications, preferences, user):
"""
Creates missing account level preferences for notifications
"""
preferences = list(preferences)
notification_types = list(set(notification.notification_type for notification in notifications))
missing_prefs = []
for notification_type in notification_types:
if not any(preference.type == notification_type for preference in preferences):
type_pref = COURSE_NOTIFICATION_TYPES.get(notification_type, {})
app_name = type_pref["notification_app"]
if type_pref.get('is_core', False):
app_pref = COURSE_NOTIFICATION_APPS.get(app_name, {})
default_pref = {
"web": app_pref["core_web"],
"push": app_pref["core_push"],
"email": app_pref["core_email"],
"email_cadence": app_pref["core_email_cadence"]
}
else:
default_pref = COURSE_NOTIFICATION_TYPES.get(notification_type, {})
missing_prefs.append(
NotificationPreference(
user=user, type=notification_type, app=app_name, web=default_pref['web'],
push=default_pref['push'], email=default_pref['email'], email_cadence=default_pref['email_cadence'],
)
)
if missing_prefs:
created_prefs = NotificationPreference.objects.bulk_create(missing_prefs, ignore_conflicts=True)
preferences = preferences + list(created_prefs)
return preferences
def filter_email_enabled_notifications(notifications, preferences, user, cadence_type=EmailCadence.DAILY):
"""
Filter notifications with email enabled in account level preferences
"""
preferences = create_missing_account_level_preferences(notifications, preferences, user)
preferences = create_account_notification_pref_if_not_exists(
[user.id],
preferences,
[notification.notification_type for notification in notifications]
)
enabled_course_prefs = [
preference.type
for preference in preferences
@@ -426,16 +365,12 @@ def update_user_preferences_from_patch(encrypted_username):
UserPreference.objects.get_or_create(user_id=user.id, key=ONE_CLICK_EMAIL_UNSUB_KEY)
def is_notification_type_channel_editable(app_name, notification_type, channel):
def is_notification_type_channel_editable(notification_type, channel):
"""
Returns if notification type channel is editable
"""
notification_type = 'core'\
if COURSE_NOTIFICATION_TYPES.get(notification_type, {}).get("is_core", False)\
else notification_type
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']
default_preferences = get_default_values_of_preferences()
return channel not in default_preferences.get(notification_type, {}).get('non_editable', [])
def get_translated_app_title(name):

82
openedx/core/djangoapps/notifications/models.py
View File

@@ -10,10 +10,8 @@ from opaque_keys.edx.django.models import CourseKeyField
from openedx.core.djangoapps.notifications.base_notification import (
get_notification_content,
COURSE_NOTIFICATION_APPS,
COURSE_NOTIFICATION_TYPES
COURSE_NOTIFICATION_TYPES, get_default_values_of_preferences
)
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
User = get_user_model()
log = logging.getLogger(__name__)
@@ -37,41 +35,6 @@ def get_additional_notification_channel_settings():
return ADDITIONAL_NOTIFICATION_CHANNEL_SETTINGS
def create_notification_preference(user_id: int, notification_type: str):
"""
Create a single notification preference with appropriate defaults.
Args:
user_id: ID of the user
notification_type: Type of notification
Returns:
NotificationPreference instance
"""
notification_config = COURSE_NOTIFICATION_TYPES.get(notification_type, {})
is_core = notification_config.get('is_core', False)
app = notification_config['notification_app']
kwargs = {
"web": notification_config.get('web', True),
"push": notification_config.get('push', False),
"email": notification_config.get('email', False),
"email_cadence": notification_config.get('email_cadence', EmailCadence.DAILY),
}
if is_core:
app_config = COURSE_NOTIFICATION_APPS[app]
kwargs = {
"web": app_config.get("core_web", True),
"push": app_config.get("core_push", False),
"email": app_config.get("core_email", False),
"email_cadence": app_config.get("core_email_cadence", EmailCadence.DAILY),
}
return NotificationPreference(
user_id=user_id,
type=notification_type,
app=app,
**kwargs,
)
class Notification(TimeStampedModel):
"""
Model to store notifications for users
@@ -143,6 +106,28 @@ class NotificationPreference(TimeStampedModel):
return f"{self.user_id} {self.type} (Web:{self.web}) (Push:{self.push})" \
f"(Email:{self.email}, {self.email_cadence})"
@property
def is_grouped(self):
"""
Returns True if the notification type is grouped.
"""
default_preference_setting = get_default_values_of_preferences().get(self.type, {})
return default_preference_setting.get('use_app_defaults', False)
@property
def config(self):
"""
Returns the configuration for the notification preference.
"""
default_preference_setting = get_default_values_of_preferences().get(self.type, {})
return {
'web': self.web,
'push': self.push,
'email': self.email,
'email_cadence': self.email_cadence,
'info': default_preference_setting.get('info', '')
}
@classmethod
def create_default_preferences_for_user(cls, user_id) -> list:
"""
@@ -189,3 +174,24 @@ class NotificationPreference(TimeStampedModel):
Returns the email cadence for the notification type.
"""
return self.email_cadence
def create_notification_preference(user_id: int, notification_type: str) -> NotificationPreference:
"""
Create a single notification preference with appropriate defaults.
Args:
user_id: ID of the user
notification_type: Type of notification
Returns:
NotificationPreference instance
"""
default_preference_setting = get_default_values_of_preferences()[notification_type]
return NotificationPreference(
user_id=user_id,
type=notification_type,
app=default_preference_setting['notification_app'],
web=default_preference_setting['web'],
push=default_preference_setting['push'],
email=default_preference_setting['email'],
email_cadence=default_preference_setting['email_cadence']
)

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

@@ -48,7 +48,7 @@ def add_info_to_notification_config(config_obj):
}
For each notification type:
- If the type is 'core', its info is fetched from `COURSE_NOTIFICATION_APPS[notification_app]['core_info']`.
- If the type is 'core', its info is fetched from `COURSE_NOTIFICATION_APPS[notification_app]['info']`.
- For all other types, info is fetched from `COURSE_NOTIFICATION_TYPES[notification_type]['info']`.
Parameters:
@@ -63,7 +63,7 @@ def add_info_to_notification_config(config_obj):
notification_types = app_prefs.get('notification_types', {})
for notification_type, type_prefs in notification_types.items():
if notification_type == "core":
type_info = COURSE_NOTIFICATION_APPS.get(notification_app, {}).get('core_info', '')
type_info = COURSE_NOTIFICATION_APPS.get(notification_app, {}).get('info', '')
else:
type_info = COURSE_NOTIFICATION_TYPES.get(notification_type, {}).get('info', '')
type_prefs['info'] = type_info
@@ -209,13 +209,16 @@ class UserNotificationPreferenceUpdateAllSerializer(serializers.Serializer):
})
# Validate notification type
if all([not COURSE_NOTIFICATION_TYPES.get(notification_type), notification_type != "core"]):
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.')
# Validate notification type and channel is editable
if notification_channel and notification_type:
if not is_notification_type_channel_editable(
notification_app,
notification_type,
"email" if notification_channel == "email_cadence" else notification_channel
):

2
openedx/core/djangoapps/notifications/settings_override.py
View File

@@ -58,5 +58,5 @@ def get_notification_apps_config() -> Dict[str, Any]:
return _apply_overrides(
default_config=DEFAULT_APPS,
setting_name='NOTIFICATION_APPS_OVERRIDE',
allowed_keys={'core_web', 'core_email', 'core_push', 'non_editable', 'core_email_cadence'}
allowed_keys={'web', 'email', 'push', 'non_editable', 'email_cadence'}
)

80
openedx/core/djangoapps/notifications/tasks.py
View File

@@ -3,7 +3,6 @@ This file contains celery tasks for notifications.
"""
import uuid
from datetime import datetime, timedelta
from typing import List
from celery import shared_task
from celery.utils.log import get_task_logger
@@ -15,7 +14,6 @@ from zoneinfo import ZoneInfo
from openedx.core.djangoapps.notifications.audience_filters import NotificationFilter
from openedx.core.djangoapps.notifications.base_notification import (
COURSE_NOTIFICATION_APPS,
COURSE_NOTIFICATION_TYPES,
get_default_values_of_preference,
get_notification_content
@@ -35,10 +33,14 @@ from openedx.core.djangoapps.notifications.grouping_notifications import (
)
from openedx.core.djangoapps.notifications.models import (
Notification,
NotificationPreference,
NotificationPreference, create_notification_preference,
)
from openedx.core.djangoapps.notifications.push.tasks import send_ace_msg_to_push_channel
from openedx.core.djangoapps.notifications.utils import clean_arguments, get_list_in_batches
from openedx.core.djangoapps.notifications.utils import (
clean_arguments,
get_list_in_batches,
create_account_notification_pref_if_not_exists
)
logger = get_task_logger(__name__)
@@ -145,7 +147,7 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
preferences = list(preferences)
if default_web_config:
preferences = create_account_notification_pref_if_not_exists(
batch_user_ids, preferences, notification_type
batch_user_ids, preferences, [notification_type]
)
if not preferences:
@@ -263,71 +265,3 @@ def update_account_user_preference(user_id: int) -> None:
# Bulk create all new preferences
NotificationPreference.objects.bulk_create(new_preferences)
return
def create_notification_preference(user_id: int, notification_type: str) -> NotificationPreference:
"""
Create a single notification preference with appropriate defaults.
Args:
user_id: ID of the user
notification_type: Type of notification
Returns:
NotificationPreference instance
"""
notification_config = COURSE_NOTIFICATION_TYPES.get(notification_type, {})
is_core = notification_config.get('is_core', False)
app = COURSE_NOTIFICATION_TYPES[notification_type]['notification_app']
email_cadence = notification_config.get('email_cadence', EmailCadence.DAILY)
if is_core:
email_cadence = COURSE_NOTIFICATION_APPS[app]['core_email_cadence']
return NotificationPreference(
user_id=user_id,
type=notification_type,
app=app,
web=_get_channel_default(is_core, notification_type, 'web'),
push=_get_channel_default(is_core, notification_type, 'push'),
email=_get_channel_default(is_core, notification_type, 'email'),
email_cadence=email_cadence,
)
def _get_channel_default(is_core: bool, notification_type: str, channel: str) -> bool:
"""
Get the default value for a notification channel.
Args:
is_core: Whether this is a core notification
notification_type: Type of notification
channel: Channel name (web, push, email)
Returns:
Default boolean value for the channel
"""
if is_core:
notification_app = COURSE_NOTIFICATION_TYPES[notification_type]['notification_app']
return COURSE_NOTIFICATION_APPS[notification_app][f'core_{channel}']
return COURSE_NOTIFICATION_TYPES[notification_type][channel]
def create_account_notification_pref_if_not_exists(user_ids: List, preferences: List, notification_type: str):
"""
Create account level notification preference if not exist.
"""
new_preferences = []
for user_id in user_ids:
if not any(preference.user_id == int(user_id) for preference in preferences):
new_preferences.append(create_notification_preference(
user_id=int(user_id),
notification_type=notification_type,
))
if new_preferences:
# ignoring conflicts because it is possible that preference is already created by another process
# conflicts may arise because of constraint on user_id and course_id fields in model
NotificationPreference.objects.bulk_create(new_preferences, ignore_conflicts=True)
preferences = preferences + new_preferences
return preferences

18
openedx/core/djangoapps/notifications/tests/test_base_notification.py
View File

@@ -15,13 +15,13 @@ class NotificationPreferenceValidationTest(ModuleStoreTestCase):
Tests if COURSE_NOTIFICATION_APPS constant has all required keys with valid
data type for new notification app
"""
bool_keys = ['enabled', 'core_web', 'core_push', 'core_email']
bool_keys = ['enabled', 'web', 'push', 'email']
notification_apps = base_notification.COURSE_NOTIFICATION_APPS
assert "" not in notification_apps
for app_data in notification_apps.values():
assert 'core_info' in app_data.keys()
assert 'info' in app_data.keys()
assert isinstance(app_data['non_editable'], list)
assert isinstance(app_data['core_email_cadence'], str)
assert isinstance(app_data['email_cadence'], str)
for key in bool_keys:
assert isinstance(app_data[key], bool)
@@ -30,13 +30,13 @@ class NotificationPreferenceValidationTest(ModuleStoreTestCase):
Tests if COURSE_NOTIFICATION_TYPES constant has all required keys with valid
data type for core notification type
"""
str_keys = ['notification_app', 'name', 'email_template']
str_keys = ['notification_app', 'name']
notification_types = base_notification.COURSE_NOTIFICATION_TYPES
assert "" not in notification_types
for notification_type in notification_types.values():
if not notification_type['is_core']:
if not notification_type.get('use_app_defaults', False):
continue
assert isinstance(notification_type['is_core'], bool)
assert isinstance(notification_type['use_app_defaults'], bool)
assert isinstance(notification_type['content_context'], dict)
assert 'content_template' in notification_type.keys()
for key in str_keys:
@@ -47,12 +47,12 @@ class NotificationPreferenceValidationTest(ModuleStoreTestCase):
Tests if COURSE_NOTIFICATION_TYPES constant has all required keys with valid
data type for non-core notification type
"""
str_keys = ['notification_app', 'name', 'info', 'email_template']
bool_keys = ['is_core', 'web', 'email', 'push']
str_keys = ['notification_app', 'name', 'info']
bool_keys = ['web', 'email', 'push']
notification_types = base_notification.COURSE_NOTIFICATION_TYPES
assert "" not in notification_types
for notification_type in notification_types.values():
if notification_type['is_core']:
if notification_type.get('use_app_defaults', False):
continue
assert 'content_template' in notification_type.keys()
assert isinstance(notification_type['content_context'], dict)

33
openedx/core/djangoapps/notifications/tests/test_settings_override.py
View File

@@ -19,29 +19,28 @@ class SettingsOverrideIntegrationTest(TestCase):
"""
@override_settings(NOTIFICATION_TYPES_OVERRIDE={
'new_discussion_post': {
'new_comment_on_response': {
'email': True,
'email_cadence': 'immediately',
'is_core': True
'use_app_defaults': False
}
})
def test_override_notification_types_real_config(self):
"""
Test overriding 'new_discussion_post' which exists in the real config.
We verify that allowed keys change and forbidden keys (is_core) do not.
Test overriding 'new_comment_on_response' which exists in the real config.
We verify that allowed keys change and forbidden keys (use_app_defaults) do not.
"""
config = get_notification_types_config()
target_notification = config['new_discussion_post']
target_notification = config['new_comment_on_response']
self.assertTrue(
target_notification['email'],
"The 'email' setting should be overridden to True."
)
self.assertFalse(
target_notification['is_core'],
"The 'is_core' field should not be overridable via settings."
self.assertTrue(
target_notification['use_app_defaults'],
"The 'use_app_defaults' field should not be overridable via settings."
)
# IMMUTABILITY CHECK: Ensure the global module variable wasn't touched
@@ -63,7 +62,7 @@ class SettingsOverrideIntegrationTest(TestCase):
@override_settings(NOTIFICATION_APPS_OVERRIDE={
'discussion': {
'core_email': False,
'email': False,
'enabled': False
}
})
@@ -76,8 +75,8 @@ class SettingsOverrideIntegrationTest(TestCase):
target_app = config['discussion']
self.assertFalse(
target_app['core_email'],
"The 'core_email' setting should be overridden to False."
target_app['email'],
"The 'email' setting should be overridden to False."
)
self.assertTrue(
@@ -86,7 +85,7 @@ class SettingsOverrideIntegrationTest(TestCase):
)
self.assertTrue(
_COURSE_NOTIFICATION_APPS['discussion']['core_email'],
_COURSE_NOTIFICATION_APPS['discussion']['email'],
"The original global _COURSE_NOTIFICATION_APPS must remain immutable."
)
@@ -128,19 +127,19 @@ class SettingsOverrideIntegrationTest(TestCase):
@override_settings(NOTIFICATION_APPS_OVERRIDE={
'discussion': {
'core_email_cadence': 'Immediately'
'email_cadence': 'Immediately'
}
})
def test_override_notification_apps_email_cadence(self):
"""
Test overriding core_email_cadence for an existing notification app.
Test overriding email_cadence for an existing notification app.
Ensures the override is applied and the module-level default isn't mutated.
"""
config = get_notification_apps_config()
target_app = config['discussion']
self.assertEqual(
target_app.get('core_email_cadence'),
target_app.get('email_cadence'),
'Immediately',
"The 'core_email_cadence' setting should be overridden to 'Immediately'."
"The 'email_cadence' setting should be overridden to 'Immediately'."
)

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

@@ -38,7 +38,8 @@ from openedx.core.djangoapps.notifications.serializers import (
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, NotificationTypeManager, COURSE_NOTIFICATION_TYPES, \
get_default_values_of_preferences
from ..utils import get_notification_types_with_visibility_settings, exclude_inaccessible_preferences
User = get_user_model()
@@ -928,3 +929,344 @@ class TestNotificationPreferencesView(ModuleStoreTestCase):
user__id=self.user.id
)
self.assertEqual(preference.email_cadence, 'Weekly')
@ddt.ddt
class TestNotificationPreferencesViewV3(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,
"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": ""
},
"content_reported": {
"web": True,
"email": True,
"push": False,
"email_cadence": "Daily",
"info": ""
},
"new_instructor_all_learners_post": {
"web": True,
"email": True,
"push": False,
"email_cadence": "Daily",
"info": ""
},
"grouped_notification": {
"web": True,
"email": True,
"push": True,
"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"],
"content_reported": ["push"],
"new_instructor_all_learners_post": ["push"]
}
},
"updates": {
"enabled": True,
"notification_types": {
"course_updates": {
"web": True,
"email": True,
"push": False,
"email_cadence": "Daily",
"info": ""
},
},
"non_editable": {
"course_updates": ["push"],
}
},
"grading": {
"enabled": True,
"notification_types": {
"ora_staff_notifications": {
"web": True,
"email": False,
"push": False,
"email_cadence": "Daily",
'info': 'Notifications for when a submission is made for ORA that includes staff '
'grading step.'
},
"ora_grade_assigned": {
"web": True,
"email": True,
"push": False,
"email_cadence": "Daily",
"info": ""
}
},
"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-v3')
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)
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,
"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": ""
},
"grouped_notification": {
"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,
"notification_types": {
"course_updates": {
"web": False,
"email": False,
"push": False,
"email_cadence": "Daily",
"info": ""
},
},
"non_editable": {
"course_updates": ["push"],
}
},
"grading": {
"enabled": True,
"notification_types": {
"ora_grade_assigned": {
"web": False,
"email": False,
"push": False,
"email_cadence": "Daily",
"info": ""
},
},
"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_grouped(self):
"""
Test case: Update notification preferences for the authenticated user
"""
update_data = {
"notification_app": "discussion",
"notification_type": "grouped_notification",
"notification_channel": "email_cadence",
"email_cadence": "Weekly"
}
grouped_types = [
key
for key, type_config in get_default_values_of_preferences().items()
if type_config.get('use_app_defaults')
]
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=grouped_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_grouped_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')

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

@@ -11,6 +11,7 @@ from .views import (
NotificationReadAPIView,
preference_update_from_encrypted_username_view,
NotificationPreferencesView,
NotificationPreferencesViewV3,
)
router = routers.DefaultRouter()
@@ -21,6 +22,11 @@ urlpatterns = [
NotificationPreferencesView.as_view(),
name='notification-preferences-aggregated-v2'
),
path(
'v3/configurations/',
NotificationPreferencesViewV3.as_view(),
name='notification-preferences-aggregated-v3'
),
path('', NotificationListAPIView.as_view(), name='notifications-list'),
path('count/', NotificationCountView.as_view(), name='notifications-count'),
path(

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

@@ -1,11 +1,12 @@
"""
Utils function for notifications app
"""
from typing import Dict, List, Set
from typing import Dict, List
from common.djangoapps.student.models import CourseAccessRole
from openedx.core.djangoapps.django_comment_common.models import Role
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx.core.djangoapps.notifications.models import create_notification_preference, NotificationPreference
from openedx.core.lib.cache_utils import request_cached
@@ -25,17 +26,6 @@ def get_list_in_batches(input_list, batch_size):
yield input_list[index: index + batch_size]
def get_user_forum_roles(user_id: int, course_id: str) -> List[str]:
"""
Get forum roles for the given user in the specified course.
:param user_id: User ID
:param course_id: Course ID
:return: List of forum roles
"""
return list(Role.objects.filter(course_id=course_id, users__id=user_id).values_list('name', flat=True))
@request_cached()
def get_notification_types_with_visibility_settings() -> Dict[str, List[str]]:
"""
@@ -85,32 +75,6 @@ def filter_out_visible_notifications(
return user_preferences
def remove_preferences_with_no_access(preferences: dict, user) -> dict:
"""
Filter out notifications visible to forum roles from user preferences.
:param preferences: User preferences dictionary
:param user: User object
:return: Updated user preferences dictionary
"""
user_preferences = preferences['notification_preference_config']
user_forum_roles = get_user_forum_roles(user.id, preferences['course_id'])
notifications_with_visibility_settings = get_notification_types_with_visibility_settings()
user_course_roles = CourseAccessRole.objects.filter(
user=user,
course_id=preferences['course_id']
).values_list('role', flat=True)
user_preferences = filter_out_visible_notifications(
user_preferences,
notifications_with_visibility_settings,
user_forum_roles,
user_course_roles
)
return preferences
def clean_arguments(kwargs):
"""
Returns query arguments from command line arguments
@@ -124,79 +88,6 @@ def clean_arguments(kwargs):
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 not target_config.get("email_cadence") or isinstance(target_config.get("email_cadence"), str):
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 get_user_forum_access_roles(user_id: int) -> List[str]:
"""
Get forum roles for the given user in all course.
@@ -229,3 +120,54 @@ def exclude_inaccessible_preferences(user_preferences: dict, user):
course_roles
)
return user_preferences
def _get_missing_preference_objects(user_ids, existing_prefs, target_types):
"""
Compares existing data against target needs and returns
a list of unsaved model instances.
"""
already_exists = {f"{p.user_id}-{p.type}" for p in existing_prefs}
to_create = []
for user_id in user_ids:
for n_type in target_types:
key = f"{int(user_id)}-{n_type}"
if key not in already_exists:
new_obj = create_notification_preference(
user_id=int(user_id),
notification_type=n_type
)
to_create.append(new_obj)
already_exists.add(key)
return to_create
def create_account_notification_pref_if_not_exists(user_ids, existing_preferences, notification_types):
"""
Ensures that NotificationPreference objects exist for the given user IDs
and notification types. Creates any missing preferences in bulk.
Args:
user_ids: Iterable of user IDs to check/create preferences for.
existing_preferences: QuerySet of existing NotificationPreference objects.
notification_types: Iterable of notification type strings to ensure exist.
Returns:
List of NotificationPreference objects including both existing and newly created ones.
"""
new_prefs_to_save = _get_missing_preference_objects(
user_ids,
existing_preferences,
notification_types
)
if not new_prefs_to_save:
return list(existing_preferences)
NotificationPreference.objects.bulk_create(
new_prefs_to_save,
ignore_conflicts=True
)
return list(existing_preferences) + new_prefs_to_save

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

@@ -20,7 +20,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
NotificationTypeManager, filter_notification_types_by_app
from .events import (
notification_preference_update_event,
notification_read_event,
@@ -37,7 +37,7 @@ from .serializers import (
from .tasks import create_notification_preference
from .utils import (
get_show_notifications_tray,
exclude_inaccessible_preferences
exclude_inaccessible_preferences, create_account_notification_pref_if_not_exists
)
@@ -446,3 +446,173 @@ class NotificationPreferencesView(APIView):
'app': validated_data['notification_app'],
}
}
@allow_any_authenticated_user()
class NotificationPreferencesViewV3(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)
# Ensure all notification types are present in the user's preferences.
user_preferences_qs = create_account_notification_pref_if_not_exists(
user_ids=[request.user.id],
existing_preferences=user_preferences_qs,
notification_types=COURSE_NOTIFICATION_TYPES.keys()
)
structured_preferences = {
app_name: {
'notification_types': {},
'enabled': COURSE_NOTIFICATION_APPS[app_name].get('enabled', True),
'non_editable': []
} for app_name in COURSE_NOTIFICATION_APPS.keys()}
for user_preference in user_preferences_qs:
app_name = user_preference.app
type_name = user_preference.type
if user_preference.is_grouped:
structured_preferences[app_name]['notification_types']['grouped_notification'] = {
**user_preference.config
}
continue
structured_preferences[app_name]['notification_types'][type_name] = {**user_preference.config}
exclude_inaccessible_preferences(structured_preferences, request.user)
structured_preferences = add_non_editable_in_preference(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'] == '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:
# 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'],
}
}