openedx.core.djangoapps.notifications package

Contents

openedx.core.djangoapps.notifications package#

Subpackages#

Submodules#

openedx.core.djangoapps.notifications.apps module#

Config for notifications app

class openedx.core.djangoapps.notifications.apps.NotificationsConfig(app_name, app_module)#

Bases: AppConfig

Config for notifications app

name = 'openedx.core.djangoapps.notifications'#
ready()#

Import signals

verbose_name = 'Notifications'#

openedx.core.djangoapps.notifications.audience_filters module#

Audience based filters for notifications and Notification filters

class openedx.core.djangoapps.notifications.audience_filters.CohortAudienceFilter(course_key)#

Bases: NotificationAudienceFilterBase

Filter class for cohort roles

filter(group_ids)#

Filter users based on their cohort ids

class openedx.core.djangoapps.notifications.audience_filters.CourseRoleAudienceFilter(course_key)#

Bases: NotificationAudienceFilterBase

Filter class for course roles

allowed_filters = ['staff', 'instructor']#
filter(course_roles)#

Filter users based on their course roles

class openedx.core.djangoapps.notifications.audience_filters.EnrollmentAudienceFilter(course_key)#

Bases: NotificationAudienceFilterBase

Filter class for enrollment modes

allowed_filters = ['audit', 'credit', 'honor', 'no-id-professional', 'professional', 'verified', 'masters', 'executive-education', 'paid-executive-education', 'unpaid-executive-education', 'paid-bootcamp', 'unpaid-bootcamp']#
filter(enrollment_modes)#

Filter users based on their course enrollment modes

class openedx.core.djangoapps.notifications.audience_filters.ForumRoleAudienceFilter(course_key)#

Bases: NotificationAudienceFilterBase

Filter class for roles

allowed_filters = ['Administrator', 'Moderator', 'Group Moderator', 'Community TA', 'Student']#
filter(roles)#

Filter users based on their roles

class openedx.core.djangoapps.notifications.audience_filters.NotificationAudienceFilterBase(course_key)#

Bases: object

Base class for notification audience filters

allowed_filters = []#
abstractmethod filter(values)#
is_valid_filter(values)#
class openedx.core.djangoapps.notifications.audience_filters.NotificationFilter#

Bases: object

Filter notifications based on their type

apply_filters(user_ids, course_key, notification_type) → list#

Apply all the filters

filter_audit_expired_users_with_no_role(user_ids, course) → list#

Check if the user has access to the course this would be true if the user has a course role or a forum role

static get_users_with_course_role(user_ids: List[int], course_id: str) → List[int]#

Get users with a course role for the given course.

static get_users_with_forum_roles(user_ids: List[int], course_id: str) → List[int]#

Get users with forum roles for the given course.

class openedx.core.djangoapps.notifications.audience_filters.TeamAudienceFilter(course_key)#

Bases: NotificationAudienceFilterBase

Filter class for team roles

filter(team_ids)#

Filter users based on team id

openedx.core.djangoapps.notifications.base_notification module#

Base setup for Notification Apps and Types.

class openedx.core.djangoapps.notifications.base_notification.NotificationApp#

Bases: TypedDict

Define the fields for values in COURSE_NOTIFICATION_APPS

An instance of this type describes a notification app, which is a way of grouping configuration of types of notifications for users.

Each notification type defined in COURSE_NOTIFICATION_TYPES also references an app.

In this case, the delivery preferences for that notification are taken

email: bool#
email_cadence: Literal['Daily', 'Weekly', 'Immediately', 'Never']#
enabled: bool#
info: str#
non_editable: list[Literal['web', 'email', 'push']]#
push: bool#
web: bool#
class openedx.core.djangoapps.notifications.base_notification.NotificationType#

Bases: TypedDict

Define the fields for values in COURSE_NOTIFICATION_TYPES

content_context: dict[str, Any]#
content_template: str#
email: NotRequired[bool]#
email_cadence: NotRequired[Literal['Daily', 'Weekly', 'Immediately', 'Never']]#
filters: list[str]#
info: NotRequired[str]#
name: str#
non_editable: NotRequired[list[Literal['web', 'email', 'push']]]#
notification_app: str#
push: NotRequired[bool]#
use_app_defaults: bool#
web: NotRequired[bool]#
openedx.core.djangoapps.notifications.base_notification.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.

Parameters:

  • 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:

Filtered dictionary containing only matching notification types

Return type:

dict

openedx.core.djangoapps.notifications.base_notification.get_default_values_of_preferences() → dict[str, dict[str, Any]]#

Returns default preferences for all notification apps

openedx.core.djangoapps.notifications.base_notification.get_notification_content(notification_type: str, context: dict[str, Any])#

Returns notification content for the given notification type with provided context.

Args: notification_type (str): The type of notification (e.g., ‘course_update’). context (dict): The context data to be used in the notification template.

Returns: str: Rendered notification content based on the template and context.

openedx.core.djangoapps.notifications.email_notifications module#

Email notifications module.

class openedx.core.djangoapps.notifications.email_notifications.EmailCadence#

Bases: object

Email cadence class

DAILY = 'Daily'#
EMAIL_CADENCE_CHOICES = [('Daily', 'Daily'), ('Weekly', 'Weekly'), ('Immediately', 'Immediately'), ('Never', 'Never')]#
EMAIL_CADENCE_CHOICES_DICT = {'Daily': 'Daily', 'Immediately': 'Immediately', 'Never': 'Never', 'Weekly': 'Weekly'}#
IMMEDIATELY = 'Immediately'#
NEVER = 'Never'#
WEEKLY = 'Weekly'#
classmethod get_email_cadence_choices()#

Returns email cadence choices.

classmethod get_email_cadence_value(email_cadence)#

Returns email cadence display for the given email cadence.

openedx.core.djangoapps.notifications.events module#

Events for notification app.

openedx.core.djangoapps.notifications.events.get_user_course_roles(user, course_id)#

Get the user’s roles in the course.

openedx.core.djangoapps.notifications.events.get_user_forums_roles(user, course_id)#

Get the user’s roles in the course forums.

openedx.core.djangoapps.notifications.events.notification_event_context(user, course_id, notification)#
openedx.core.djangoapps.notifications.events.notification_generated_event(user_ids, app_name, notification_type, course_key, content_url, content, sender_id=None)#

Emit an event when a notification is generated.

openedx.core.djangoapps.notifications.events.notification_preference_unsubscribe_event(user, is_preference_updated=False)#

Emits an event when user clicks on one-click-unsubscribe url

openedx.core.djangoapps.notifications.events.notification_preference_update_event(user, course_id, updated_preference)#

Emit an event when a notification preference is updated.

openedx.core.djangoapps.notifications.events.notification_read_event(user, notification, first_read=False)#

Emit an event when a notification app is marked read for a user.

openedx.core.djangoapps.notifications.events.notification_tray_opened_event(user, unseen_notifications_count)#

Emit an event when a notification tray is opened.

openedx.core.djangoapps.notifications.events.notifications_app_all_read_event(user, app_name)#

Emit an event when a notification is read.

openedx.core.djangoapps.notifications.exceptions module#

Notification-related exceptions.

exception openedx.core.djangoapps.notifications.exceptions.InvalidNotificationTypeError#

Bases: Exception

Exception raised when an invalid notification type is passed.

openedx.core.djangoapps.notifications.grouping_notifications module#

Notification grouping utilities for notifications

class openedx.core.djangoapps.notifications.grouping_notifications.BaseNotificationGrouper#

Bases: ABC

Base class for notification groupers.

abstractmethod group(new_notification, old_notification)#
class openedx.core.djangoapps.notifications.grouping_notifications.NewPostGrouper#

Bases: BaseNotificationGrouper

Groups new post notifications based on the author name.

group(new_notification, old_notification)#

Groups new post notifications based on the author name.

class openedx.core.djangoapps.notifications.grouping_notifications.NewResponseGrouper#

Bases: BaseNotificationGrouper

Grouper for new response on post.

group(new_notification, old_notification)#

Groups new ora staff notifications based on the xblock ID.

class openedx.core.djangoapps.notifications.grouping_notifications.NewResponseOnFollowedPostGrouper#

Bases: BaseNotificationGrouper

Grouper for new response on post.

group(new_notification, old_notification)#

Groups new ora staff notifications based on the xblock ID.

class openedx.core.djangoapps.notifications.grouping_notifications.NotificationRegistry#

Bases: object

Registry for notification groupers.

classmethod get_grouper(notification_type: str) → BaseNotificationGrouper | None#

Retrieves the appropriate notification grouper based on the given notification type.

Parameters:

notification_type – The type of notification for which to retrieve the grouper.

Returns:

The corresponding BaseNotificationGrouper instance or None if no grouper is found.

classmethod register(notification_type: str)#

Registers a notification grouper for the given notification type. Args

notification_type: The type of notification for which to register the grouper.

Returns:

A decorator that registers the grouper class for the given notification type.

class openedx.core.djangoapps.notifications.grouping_notifications.OraStaffGrouper#

Bases: BaseNotificationGrouper

Grouper for new ora staff notifications.

group(new_notification, old_notification)#

Groups new ora staff notifications based on the xblock ID.

openedx.core.djangoapps.notifications.grouping_notifications.get_user_existing_notifications(user_ids, notification_type, group_by_id, course_id)#

Returns user last group able notification

openedx.core.djangoapps.notifications.grouping_notifications.group_user_notifications(new_notification: Notification, old_notification: Notification)#

Groups user notification based on notification type and group_id

openedx.core.djangoapps.notifications.handlers module#

Handlers for notifications

openedx.core.djangoapps.notifications.handlers.calculate_course_wide_notification_audience(course_key, audience_filters)#

Calculate the audience for a course-wide notification based on the audience filters

openedx.core.djangoapps.notifications.handlers.create_user_account_preferences(sender, instance, created, **kwargs)#

Initialize user notification preferences when new user is created.

openedx.core.djangoapps.notifications.handlers.generate_course_notifications(signal, sender, course_notification_data, metadata, **kwargs)#

Watches for COURSE_NOTIFICATION_REQUESTED signal and calls send_notifications task

openedx.core.djangoapps.notifications.handlers.generate_user_notifications(signal, sender, notification_data, metadata, **kwargs)#

Watches for USER_NOTIFICATION_REQUESTED signal and calls send_web_notifications task

openedx.core.djangoapps.notifications.models module#

Models for notifications

class openedx.core.djangoapps.notifications.models.DigestSchedule(*args, **kwargs)#

Bases: TimeStampedModel

Tracks scheduled Celery digest tasks for daily/weekly email digests.

One record exists per (user, cadence_type, delivery_time) combination, representing a single pending Celery task. This is the source of truth for deduplication of digest scheduling.

NOTE: This is intentionally separate from Notification.email_scheduled. Notification.email_scheduled serves the immediate/buffer cadence flow (decide_email_action → schedule_digest_buffer → send_buffered_digest) and operates at the notification row level. DigestSchedule operates at the task level — one record per scheduled Celery job — for daily/weekly digests only.

exception DoesNotExist#

Bases: ObjectDoesNotExist

exception MultipleObjectsReturned#

Bases: MultipleObjectsReturned

cadence_type#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

created#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

delivery_time#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_next_by_created(*, field=<model_utils.fields.AutoCreatedField: created>, is_next=True, **kwargs)#
get_next_by_delivery_time(*, field=<django.db.models.fields.DateTimeField: delivery_time>, is_next=True, **kwargs)#
get_next_by_modified(*, field=<model_utils.fields.AutoLastModifiedField: modified>, is_next=True, **kwargs)#
get_previous_by_created(*, field=<model_utils.fields.AutoCreatedField: created>, is_next=False, **kwargs)#
get_previous_by_delivery_time(*, field=<django.db.models.fields.DateTimeField: delivery_time>, is_next=False, **kwargs)#
get_previous_by_modified(*, field=<model_utils.fields.AutoLastModifiedField: modified>, is_next=False, **kwargs)#
id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

modified#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects = <django.db.models.manager.Manager object>#
task_id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

user#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

user_id#
class openedx.core.djangoapps.notifications.models.Notification(*args, **kwargs)#

Bases: TimeStampedModel

Model to store notifications for users

exception DoesNotExist#

Bases: ObjectDoesNotExist

exception MultipleObjectsReturned#

Bases: MultipleObjectsReturned

app_name#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

property content#

Returns the content for the notification.

content_context#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

content_url#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

course_id#

DO NOT REUSE THIS CLASS. Provided for backwards compatibility only!

A placeholder class that provides a way to set the attribute on the model.

created#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

email#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

email_scheduled#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

email_sent_on#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_next_by_created(*, field=<model_utils.fields.AutoCreatedField: created>, is_next=True, **kwargs)#
get_next_by_modified(*, field=<model_utils.fields.AutoLastModifiedField: modified>, is_next=True, **kwargs)#
get_previous_by_created(*, field=<model_utils.fields.AutoCreatedField: created>, is_next=False, **kwargs)#
get_previous_by_modified(*, field=<model_utils.fields.AutoLastModifiedField: modified>, is_next=False, **kwargs)#
group_by_id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

last_read#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

last_seen#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

modified#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

notification_type#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects = <django.db.models.manager.Manager object>#
push#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

user#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

user_id#
web#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

class openedx.core.djangoapps.notifications.models.NotificationPreference(*args, **kwargs)#

Bases: TimeStampedModel

Model to store notification preferences for users at account level

exception DoesNotExist#

Bases: ObjectDoesNotExist

class EmailCadenceChoices(*values)#

Bases: TextChoices

DAILY = 'Daily'#
IMMEDIATELY = 'Immediately'#
WEEKLY = 'Weekly'#
exception MultipleObjectsReturned#

Bases: MultipleObjectsReturned

app#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

property config#

Returns the configuration for the notification preference.

classmethod create_default_preferences_for_user(user_id) → list#

Creates all preferences for user Note: It creates preferences using bulk create, so primary key will be missing for newly created preference. Refetch if primary key is needed

created#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

email#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

email_cadence#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_channels_for_notification_type(*args, **kwargs) → list#

Returns the channels for the given app name and notification type. Sample Response: [‘web’, ‘push’]

get_email_cadence_display(*, field=<django.db.models.fields.CharField: email_cadence>)#
get_email_cadence_for_notification_type(*args, **kwargs) → str#

Returns the email cadence for the notification type.

get_next_by_created(*, field=<model_utils.fields.AutoCreatedField: created>, is_next=True, **kwargs)#
get_next_by_modified(*, field=<model_utils.fields.AutoLastModifiedField: modified>, is_next=True, **kwargs)#
get_previous_by_created(*, field=<model_utils.fields.AutoCreatedField: created>, is_next=False, **kwargs)#
get_previous_by_modified(*, field=<model_utils.fields.AutoLastModifiedField: modified>, is_next=False, **kwargs)#
id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

is_active#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

is_enabled_for_any_channel(*args, **kwargs) → bool#

Returns True if the notification preference is enabled for any channel.

property is_grouped#

Returns True if the notification type is grouped.

modified#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects = <django.db.models.manager.Manager object>#
push#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

type#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

user#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

user_id#
web#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

openedx.core.djangoapps.notifications.models.create_notification_preference(user_id: int, notification_type: str) → NotificationPreference#

Create a single notification preference with appropriate defaults. :param user_id: ID of the user :param notification_type: Type of notification

Returns:

NotificationPreference instance

openedx.core.djangoapps.notifications.models.get_additional_notification_channel_settings()#

Returns the additional notification channel settings.

openedx.core.djangoapps.notifications.models.get_notification_channels()#

Returns the notification channels.

openedx.core.djangoapps.notifications.notification_content module#

Helper functions for overriding notification content for given notification type.

openedx.core.djangoapps.notifications.notification_content.get_comment_on_followed_post_notification_context(context: Dict) → Dict#

Returns the context for the given notification type with the author pronoun.

openedx.core.djangoapps.notifications.notification_content.get_new_comment_notification_context(context: Dict) → Dict#

Returns the context for the given notification type with the author pronoun.

openedx.core.djangoapps.notifications.notification_content.get_notification_context_with_author_pronoun(context: Dict) → Dict#

Returns the context for the given notification type with the author pronoun.

openedx.core.djangoapps.notifications.notification_content.get_notification_type_context_function(notification_type) → callable#
Returns:

The function that returns the context for the given notification type.

Return type:

callable

openedx.core.djangoapps.notifications.permissions module#

Permissions for notifications

openedx.core.djangoapps.notifications.permissions.allow_any_authenticated_user()#

Function and class decorator that abstracts the authentication and permission checks for api views. Allows both verified and non-verified users

openedx.core.djangoapps.notifications.policies module#

Policies for the notifications app.

class openedx.core.djangoapps.notifications.policies.CoursePushNotificationOptout#

Bases: Policy

Course Push Notification optOut Policy.

check(message)#

Check if the user has opted out of push notifications for the given course. :param message: :return: PolicyResult

openedx.core.djangoapps.notifications.serializers module#

Serializers for the notifications API.

class openedx.core.djangoapps.notifications.serializers.NotificationSerializer(*args, **kwargs)#

Bases: ModelSerializer

Serializer for the Notification model.

class Meta#

Bases: object

fields = ('id', 'app_name', 'notification_type', 'content_context', 'content', 'content_url', 'course_id', 'last_read', 'last_seen', 'created')#
model#

alias of Notification

class openedx.core.djangoapps.notifications.serializers.UserNotificationPreferenceUpdateAllSerializer(*args, **kwargs)#

Bases: Serializer

Serializer for user notification preferences update with custom field validators.

validate(attrs)#

Cross-field validation for notification preference update.

openedx.core.djangoapps.notifications.serializers.add_info_to_notification_config(config_obj)#

Enhances the notification configuration by appending descriptive ‘info’ to each notification type.

This function supports two different structures of config_obj, depending on the source of the data: either from the account preferences API (AggregatedNotificationPreferences) or the course preferences API (UserNotificationPreferenceView).

Supported input structures:

  1. From account preferences API:
    {
    ‘notification_app’: {
    ‘notification_types’: {

    ‘core’: { … }, ‘non-core’: { … }

    }

    }

    }

  2. From course preferences API:
    {
    ‘notification_preference_config’: {
    ‘notification_app’: {
    ‘notification_types’: {

    ‘core’: { … }, ‘non-core’: { … }

    }

    }

    }

    }

For each notification type: - 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:

config_obj (dict) – The notification configuration object to enhance.

Returns:

The enhanced configuration object with added ‘info’ fields.

Return type:

dict

openedx.core.djangoapps.notifications.serializers.add_non_editable_in_preference(preference)#

Add non_editable preferences to the preference dict

openedx.core.djangoapps.notifications.serializers.validate_email_cadence(email_cadence: str) → str#

Validate email cadence value.

openedx.core.djangoapps.notifications.serializers.validate_notification_app(notification_app: str) → str#

Validate notification app value.

openedx.core.djangoapps.notifications.serializers.validate_notification_app_enabled(notification_app: str) → str#

Validate notification app is enabled.

openedx.core.djangoapps.notifications.serializers.validate_notification_channel(notification_channel: str) → str#

Validate notification channel value.

openedx.core.djangoapps.notifications.serializers.validate_notification_type(notification_type: str) → str#

Validate notification type value.

openedx.core.djangoapps.notifications.settings_override module#

Settings override module for notification configurations.

This module provides functionality to override notification configurations via Django settings.

openedx.core.djangoapps.notifications.settings_override.get_notification_apps_config() → Dict[str, Any]#

Get COURSE_NOTIFICATION_APPS configuration with settings overrides applied.

openedx.core.djangoapps.notifications.settings_override.get_notification_types_config() → Dict[str, Any]#

Get COURSE_NOTIFICATION_TYPES configuration with settings overrides applied.

openedx.core.djangoapps.notifications.tasks module#

This file contains celery tasks for notifications.

openedx.core.djangoapps.notifications.tasks.is_notification_valid(notification_type, context)#

Validates notification before creation

openedx.core.djangoapps.notifications.tasks.update_account_user_preference(user_id: int) → None#

Update account level user preferences to ensure all notification types are present.

openedx.core.djangoapps.notifications.urls module#

URLs for the notifications API.

openedx.core.djangoapps.notifications.utils module#

Utils function for notifications app

openedx.core.djangoapps.notifications.utils.clean_arguments(kwargs)#

Returns query arguments from command line arguments

openedx.core.djangoapps.notifications.utils.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. :param user_ids: Iterable of user IDs to check/create preferences for. :param existing_preferences: QuerySet of existing NotificationPreference objects. :param notification_types: Iterable of notification type strings to ensure exist.

Returns:

List of NotificationPreference objects including both existing and newly created ones.

openedx.core.djangoapps.notifications.utils.exclude_inaccessible_preferences(user_preferences: dict, user)#

Exclude notifications from user preferences that the user has no access to, based on forum and course roles.

Parameters:

  • user_preferences – Dictionary of user notification preferences

  • user – Django User object

Returns:

Updated user_preferences dictionary (modified in-place)

openedx.core.djangoapps.notifications.utils.filter_out_visible_notifications(user_preferences: dict, notifications_with_visibility: Dict[str, List[str]], user_forum_roles: List[str], user_course_roles: List[str]) → dict#

Filter out notifications visible to forum roles from user preferences.

Parameters:

  • user_preferences – User preferences dictionary

  • notifications_with_visibility – List of dictionaries with notification type names and

corresponding visibility settings :param user_forum_roles: List of forum roles for the user :param user_course_roles: List of course roles for the user :return: Updated user preferences dictionary

openedx.core.djangoapps.notifications.utils.get_list_in_batches(input_list, batch_size)#

Divides the list of objects into list of list of objects each of length batch_size.

openedx.core.djangoapps.notifications.utils.get_notification_types_with_visibility_settings() → Dict[str, List[str]]#

Get notification types with their visibility settings.

Returns:

List of dictionaries with notification type names and corresponding visibility settings

openedx.core.djangoapps.notifications.utils.get_show_notifications_tray()#

Returns whether notifications tray is enabled via waffle flag

openedx.core.djangoapps.notifications.utils.get_user_forum_access_roles(user_id: int) → List[str]#

Get forum roles for the given user in all course.

Parameters:

user_id – User ID

Returns:

List of forum roles

openedx.core.djangoapps.notifications.views module#

Views for the notifications API.

class openedx.core.djangoapps.notifications.views.MarkNotificationsSeenAPIView(**kwargs)#

Bases: UpdateAPIView

API view for marking user’s all notifications seen for a provided app_name.

authentication_classes = (<class 'edx_rest_framework_extensions.auth.jwt.authentication.JwtAuthentication'>, <class 'openedx.core.lib.api.authentication.BearerAuthenticationAllowInactiveUser'>, <class 'edx_rest_framework_extensions.auth.session.authentication.SessionAuthenticationAllowInactiveUser'>)#
permission_classes = (<class 'rest_framework.permissions.IsAuthenticated'>,)#
update(request, *args, **kwargs)#

Marks all notifications for the given app name seen for the authenticated user.

Args:

app_name: The name of the app to mark notifications seen for.

Response Format:

A Response object with a 200 OK status code if the notifications were successfully marked seen.

Response Error Codes: - 400: Bad Request status code if the app name is invalid.

class openedx.core.djangoapps.notifications.views.NotificationCountView(**kwargs)#

Bases: APIView

API view for getting the unseen notifications count and show_notification_tray flag for a user.

authentication_classes = (<class 'edx_rest_framework_extensions.auth.jwt.authentication.JwtAuthentication'>, <class 'openedx.core.lib.api.authentication.BearerAuthenticationAllowInactiveUser'>, <class 'edx_rest_framework_extensions.auth.session.authentication.SessionAuthenticationAllowInactiveUser'>)#
get(request)#

Get the unseen notifications count and show_notification_tray flag for a user.

Permissions: User must be authenticated. Response Format: ```json {

“show_notifications_tray”: (bool) show_notifications_tray, “count”: (int) total_number_of_unseen_notifications, “count_by_app_name”: {

(str) app_name: (int) number_of_unseen_notifications, …

}, “notification_expiry_days”: 60

}#

Response Error Codes: - 403: The requester cannot access resource.

permission_classes = (<class 'rest_framework.permissions.IsAuthenticated'>,)#
class openedx.core.djangoapps.notifications.views.NotificationListAPIView(**kwargs)#

Bases: ListAPIView

API view for listing notifications for a user.

Permissions: User must be authenticated. Response Format (paginated):

{
“results”[
{

“id”: (int) notification_id, “app_name”: (str) app_name, “notification_type”: (str) notification_type, “content”: (str) content, “content_context”: (dict) content_context, “content_url”: (str) content_url, “last_read”: (datetime) last_read, “last_seen”: (datetime) last_seen

], “count”: (int) total_number_of_notifications, “next”: (str) url_to_next_page_of_notifications, “previous”: (str) url_to_previous_page_of_notifications, “page_size”: (int) number_of_notifications_per_page,

}

Response Error Codes: - 403: The requester cannot access resource.

authentication_classes = (<class 'edx_rest_framework_extensions.auth.jwt.authentication.JwtAuthentication'>, <class 'openedx.core.lib.api.authentication.BearerAuthenticationAllowInactiveUser'>, <class 'edx_rest_framework_extensions.auth.session.authentication.SessionAuthenticationAllowInactiveUser'>)#
get_queryset()#

Override the get_queryset method to filter the queryset by app name, request.user and created

permission_classes = (<class 'rest_framework.permissions.IsAuthenticated'>,)#
serializer_class#

alias of NotificationSerializer

class openedx.core.djangoapps.notifications.views.NotificationPreferencesViewV3(**kwargs)#

Bases: APIView

API view to retrieve and structure the notification preferences for the authenticated user.

authentication_classes = (<class 'edx_rest_framework_extensions.auth.jwt.authentication.JwtAuthentication'>, <class 'openedx.core.lib.api.authentication.BearerAuthenticationAllowInactiveUser'>, <class 'edx_rest_framework_extensions.auth.session.authentication.SessionAuthenticationAllowInactiveUser'>)#
get(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. This provides a complete view of all possible notifications and the user’s current settings for them.

Returns:

A DRF Response object containing the structured

notification preferences or an error message.

Return type:

Response

permission_classes = (<class 'rest_framework.permissions.IsAuthenticated'>,)#
put(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:

A DRF Response object indicating success or failure.

Return type:

Response

class openedx.core.djangoapps.notifications.views.NotificationReadAPIView(**kwargs)#

Bases: APIView

API view for marking user notifications as read, either all notifications or a single notification

authentication_classes = (<class 'edx_rest_framework_extensions.auth.jwt.authentication.JwtAuthentication'>, <class 'openedx.core.lib.api.authentication.BearerAuthenticationAllowInactiveUser'>, <class 'edx_rest_framework_extensions.auth.session.authentication.SessionAuthenticationAllowInactiveUser'>)#
patch(request, *args, **kwargs)#

Marks all notifications or single notification read for the given app name or notification id for the authenticated user.

Requests: PATCH /api/notifications/read/

Parameters:

request (Request) –

The request object containing the app name or notification id. {

”app_name”: (str) app_name, “notification_id”: (int) notification_id

}

Returns: - 200: OK status code if the notification or notifications were successfully marked read. - 400: Bad Request status code if the app name is invalid. - 403: Forbidden status code if the user is not authenticated. - 404: Not Found status code if the notification was not found.

permission_classes = (<class 'rest_framework.permissions.IsAuthenticated'>,)#
openedx.core.djangoapps.notifications.views.preference_update_from_encrypted_username_view(request, *args, **kwargs)#

View to update user preferences from encrypted username and patch. username and patch must be string

Module contents#

Contents