openedx.core.djangoapps.notifications.email package

openedx.core.djangoapps.notifications.email package#

Submodules#

openedx.core.djangoapps.notifications.email.events module#

Events for email notifications

openedx.core.djangoapps.notifications.email.events.send_immediate_email_digest_sent_event(user, cadence_type, notification)#: Sends tracker and segment event for immediate notification email

openedx.core.djangoapps.notifications.email.events.send_user_email_digest_sent_event(user, cadence_type, notifications, message_context)#: Sends tracker and segment email for user email digest

openedx.core.djangoapps.notifications.email.message_type module#

Email notifications MessageType

class openedx.core.djangoapps.notifications.email.message_type.EmailNotificationMessageType(*args, **kwargs)#

Bases: MessageType

Edx-ace MessageType for Email Notifications

NAME = 'notifications'#

openedx.core.djangoapps.notifications.email.notification_icons module#

Notification Icons

class openedx.core.djangoapps.notifications.email.notification_icons.NotificationTypeIcons#

Bases: object

Notification Mapping with icons

CHECK_CIRCLE_GREEN = 'CHECK_CIRCLE_GREEN'#

HELP_OUTLINE = 'HELP_OUTLINE'#

NEWSPAPER = 'NEWSPAPER'#

OPEN_RESPONSE_OUTLINE = 'OPEN_RESPONSE_OUTLINE'#

POST_OUTLINE = 'POST_OUTLINE'#

QUESTION_ANSWER_OUTLINE = 'QUESTION_ANSWER_OUTLINE'#

REPORT_RED = 'REPORT_RED'#

VERIFIED = 'VERIFIED'#

classmethod get_icon_name_for_notification_type(notification_type, default='POST_OUTLINE')#: Returns icon name for notification type

classmethod get_icon_url_for_notification_type(notification_type)#: Returns icon url for notification type

openedx.core.djangoapps.notifications.email.tasks module#

Celery tasks for sending email notifications

openedx.core.djangoapps.notifications.email.tasks.add_to_existing_buffer(notification: Notification) → None#

Add notification to existing buffer. Just mark as scheduled - the existing job will find it!

Called for THIRD+ notifications.

openedx.core.djangoapps.notifications.email.tasks.decide_email_action(user: User, course_key: str, notification: Notification) → str#

Decide what to do with this notification.

Logic: - No recent email? → send_immediate (1st) - Recent email + no buffer? → schedule_buffer (2nd) - Recent email + buffer exists? → add_to_buffer (3rd+)

Returns:: ‘send_immediate’, ‘schedule_buffer’, or ‘add_to_buffer’

openedx.core.djangoapps.notifications.email.tasks.get_audience_for_cadence_email(cadence_type)#: Returns users that are eligible to receive cadence email

openedx.core.djangoapps.notifications.email.tasks.get_buffer_minutes() → int#: Get configured buffer period in minutes.

openedx.core.djangoapps.notifications.email.tasks.get_digest_dedupe_key(user_id, cadence_type, delivery_time)#

Generate a deduplication key for a digest email task.

This key ensures that only one digest task is scheduled per user per cadence period.

Returns:: A unique key based on user_id, cadence_type, and delivery window.
Return type:: str

openedx.core.djangoapps.notifications.email.tasks.get_next_digest_delivery_time(cadence_type)#

Calculate the next delivery time for a digest email based on cadence type.

Uses Django settings for configurable delivery time/day: - NOTIFICATION_DAILY_DIGEST_DELIVERY_HOUR (default: 17) - NOTIFICATION_DAILY_DIGEST_DELIVERY_MINUTE (default: 0) - NOTIFICATION_WEEKLY_DIGEST_DELIVERY_DAY (default: 0 = Monday) - NOTIFICATION_WEEKLY_DIGEST_DELIVERY_HOUR (default: 17) - NOTIFICATION_WEEKLY_DIGEST_DELIVERY_MINUTE (default: 0)

Returns:: The next scheduled delivery time in UTC.
Return type:: datetime

openedx.core.djangoapps.notifications.email.tasks.is_digest_already_scheduled(user_id, cadence_type, delivery_time)#

Check if a digest email is already scheduled for this user in the current cadence window.

This prevents duplicate scheduling when multiple notifications arrive in the same digest window.

Uses DigestSchedule model for an exact (user, cadence_type, delivery_time) lookup — one record represents one pending Celery task. This is intentionally separate from Notification.email_scheduled, which tracks the immediate/buffer cadence flow and operates at the notification row level rather than the task level.

openedx.core.djangoapps.notifications.email.tasks.is_digest_already_sent_in_window(user_id, cadence_type, delivery_time)#

Check if a digest email has already been sent for this user in the current cadence window.

This prevents duplicate emails when both cron jobs and delayed tasks co-exist.

openedx.core.djangoapps.notifications.email.tasks.schedule_bulk_digest_emails(user_cadence_map)#

Bulk-schedule delayed Celery tasks for digest emails for multiple users.

This avoids N+1 query issues when scheduling digests for many users at once (e.g. during send_notifications) by batching DB operations.

Runs ~3 queries per cadence type regardless of user count: 1. SELECT existing DigestSchedule records (1 query) 2. Bulk INSERT new DigestSchedule records (1 query) 3. Bulk UPDATE Notification.email_scheduled (1 query)

Parameters:: user_cadence_map – dict mapping user_id -> cadence_type (EmailCadence.DAILY or EmailCadence.WEEKLY)

openedx.core.djangoapps.notifications.email.tasks.schedule_digest_buffer(user: User, notification: Notification, course_key: str, user_language: str) → None#: Schedule a buffer job for digest email. Called for the SECOND notification only.

openedx.core.djangoapps.notifications.email.tasks.send_digest_email_to_user(user: User, cadence_type: str, start_date: datetime, end_date: datetime, user_language: str = 'en', courses_data: dict = None)#: Send [cadence_type] email to user. Cadence Type can be EmailCadence.DAILY or EmailCadence.WEEKLY start_date: Datetime object end_date: Datetime object

openedx.core.djangoapps.notifications.email.tasks.send_immediate_cadence_email(email_notification_mapping, course_key)#

Send immediate cadence email to users :param email_notification_mapping: Dictionary of user_id and Notification object :param course_key: Course key for which the email is sent

First notification → Send immediately
Second notification → Schedule buffer job (15 min)
Third+ notifications → Just mark as scheduled (no new job)

openedx.core.djangoapps.notifications.email.tasks.send_immediate_email(user: User, notification: Notification, course_key: str, course_name: str, user_language: str) → None#: Send immediate email for the first notification.

openedx.core.djangoapps.notifications.email.utils module#

Email Notifications Utils

openedx.core.djangoapps.notifications.email.utils.add_additional_attributes_to_notifications(notifications, courses_data=None)#: Add attributes required for email content to notification instance notifications: list[Notification] course_data: Cache course info

openedx.core.djangoapps.notifications.email.utils.add_headers_to_email_message(message, context)#: Add headers to email message

openedx.core.djangoapps.notifications.email.utils.add_zero_margin_to_root(html_string)#: Adds to zero margin to root element of html string

openedx.core.djangoapps.notifications.email.utils.create_app_notifications_dict(notifications)#: Return a dictionary with notification app as key and title, count and notifications as its value

openedx.core.djangoapps.notifications.email.utils.create_datetime_string(datetime_instance)#: Returns string for datetime object

openedx.core.djangoapps.notifications.email.utils.create_email_digest_context(app_notifications_dict, username, start_date, end_date=None, digest_frequency='Daily', courses_data=None)#: Creates email context based on content app_notifications_dict: Mapping of notification app and its count, title and notifications start_date: datetime instance end_date: datetime instance digest_frequency: EmailCadence.DAILY or EmailCadence.WEEKLY courses_data: Dictionary to cache course info (avoid additional db calls)

openedx.core.djangoapps.notifications.email.utils.create_email_template_context(username)#: Creates email context for header and footer

openedx.core.djangoapps.notifications.email.utils.decrypt_string(string)#: Decrypts input string

openedx.core.djangoapps.notifications.email.utils.encrypt_string(string)#: Encrypts input string

openedx.core.djangoapps.notifications.email.utils.filter_email_enabled_notifications(notifications, preferences, user, cadence_type='Daily')#: Filter notifications with email enabled in account level preferences

openedx.core.djangoapps.notifications.email.utils.get_course_info(course_key)#: Returns course info for course_key

openedx.core.djangoapps.notifications.email.utils.get_icon_url_for_notification_type(notification_type)#: Returns icon url for notification type

openedx.core.djangoapps.notifications.email.utils.get_language_preference_for_users(user_ids)#: Returns mapping of user_id and language preference for users

openedx.core.djangoapps.notifications.email.utils.get_start_end_date(cadence_type)#: Returns start_date and end_date for email digest

openedx.core.djangoapps.notifications.email.utils.get_text_for_notification_type(notification_type)#: Returns text for notification type

openedx.core.djangoapps.notifications.email.utils.get_time_ago(datetime_obj)#: Returns time_ago for datetime instance

openedx.core.djangoapps.notifications.email.utils.get_translated_app_title(name)#: Returns translated string from notification app_name key

openedx.core.djangoapps.notifications.email.utils.get_unique_course_ids(notifications)#: Returns unique course_ids from notifications

openedx.core.djangoapps.notifications.email.utils.get_unsubscribe_link(username)#: Returns unsubscribe url for username with patch preferences

openedx.core.djangoapps.notifications.email.utils.is_notification_type_channel_editable(notification_type, channel)#: Returns if notification type channel is editable

openedx.core.djangoapps.notifications.email.utils.update_user_preferences_from_patch(encrypted_username)#: Decrypt username and patch and updates user preferences Allowed parameters for decrypted patch

app_name: name of app notification_type: notification type name channel: channel name (‘web’, ‘push’, ‘email’) value: True or False course_id: course key string

openedx.core.djangoapps.notifications.email.utils.username_from_hash(group, request)#: Django ratelimit key to return username from hash