feat: add management command to generate notification preferences for course

openedx/core/djangoapps/notifications/management/commands/generate_course_notification_preferences.py

@@ -0,0 +1,34 @@
+"""
+Management command for creating Course Notification Preferences for users in course.
+"""
+
+import logging
+
+from django.core.management.base import BaseCommand
+
+from openedx.core.djangoapps.notifications.tasks import create_course_notification_preferences_for_courses
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    Invoke with:
+
+        python manage.py [lms|cms] generate_course_notification_preferences [course_id] [course_id] ...
+    """
+    help = (
+        "Back-fill missing course notification preferences. This will queue a celery task to "
+        "create course notification preferences for all active enrollments in the courses provided."
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            'course_ids',
+            help='course_ids seperated by space for which to create Course Notification Preferences.',
+            nargs='*'
+        )
+
+    def handle(self, *args, **options):
+        course_ids = options['course_ids']
+        create_course_notification_preferences_for_courses.delay(course_ids)

openedx/core/djangoapps/notifications/management/tests/test_generate_course_notification_preferences.py

@@ -0,0 +1,32 @@
+"""
+Test for generate_notification_preferences management command.
+"""
+
+from unittest import mock
+
+from django.core.management import call_command
+
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+
+
+class GenerateCourseNotificationPreferencesTests(ModuleStoreTestCase):
+    """
+    Tests for generate_course_notification_preferences management command.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.course_ids = ['course-v1:edX+DemoX.1+2T2017', 'course-v1:edX+DemoX.1+2T2018']
+
+    @mock.patch(
+        'openedx.core.djangoapps.notifications.tasks.create_course_notification_preferences_for_courses'
+    )
+    def test_generate_course_notification_preferences(self, mock_task):
+        """
+        Test generate_course_notification_preferences command.
+        """
+        call_command(
+            'generate_course_notification_preferences',
+            self.course_ids,
+        )
+        mock_task.delay.assert_called_once_with(self.course_ids)

openedx/core/djangoapps/notifications/tasks.py

@@ -0,0 +1,40 @@
+"""
+This file contains celery tasks for notifications.
+"""
+
+from celery import shared_task
+from celery.utils.log import get_task_logger
+from django.db import transaction
+from edx_django_utils.monitoring import set_code_owner_attribute
+
+from common.djangoapps.student.models import CourseEnrollment
+from openedx.core.djangoapps.notifications.models import CourseNotificationPreference
+
+logger = get_task_logger(__name__)
+
+
+@shared_task(bind=True, ignore_result=True)
+@set_code_owner_attribute
+@transaction.atomic
+def create_course_notification_preferences_for_courses(self, course_ids):
+    """
+    This task creates Course Notification Preferences for users in courses.
+    """
+    logger.info('Running task create_course_notification_preferences')
+    newly_created = 0
+    for course_id in course_ids:
+        enrollments = CourseEnrollment.objects.filter(course_id=course_id, is_active=True)
+        logger.info(f'Found {enrollments.count()} enrollments for course {course_id}')
+        logger.info(f'Creating Course Notification Preferences for course {course_id}')
+        for enrollment in enrollments:
+            _, created = CourseNotificationPreference.objects.get_or_create(
+                user=enrollment.user, course_id=course_id
+            )
+            if created:
+                newly_created += 1
+
+        logger.info(
+            f'CourseNotificationPreference back-fill completed for course {course_id}.\n'
+            f'Newly created course preferences: {newly_created}.\n'
+        )
+    logger.info('Completed task create_course_notification_preferences')