GradingPolicyChanged Signal Handler

https://openedx.atlassian.net/browse/EDUCATOR-393

cms/celery.py

@@ -37,6 +37,5 @@ class Router(AlternateEnvironmentRouter):
         return {
             'openedx.core.djangoapps.content.block_structure.tasks.update_course_in_cache': 'lms',
             'openedx.core.djangoapps.content.block_structure.tasks.update_course_in_cache_v2': 'lms',
-            'openedx.core.djangoapps.grades.tasks.compute_grades_for_course': 'lms',
-            'openedx.core.djangoapps.grades.tasks.compute_grades_for_course_v2': 'lms',
+            'lms.djangoapps.grades.tasks.compute_all_grades_for_course': 'lms',
         }

cms/djangoapps/contentstore/apps.py

@@ -9,7 +9,7 @@ from django.apps import AppConfig
 
 class ContentstoreConfig(AppConfig):
     """
-    Application Configuration for Grades.
+    Application Configuration for Contentstore.
     """
     name = u'contentstore'
 

cms/djangoapps/contentstore/signals/handlers.py

@@ -8,11 +8,14 @@ from pytz import UTC
 
 from contentstore.courseware_index import CoursewareSearchIndexer, LibrarySearchIndexer
 from contentstore.proctoring import register_special_exams
+from lms.djangoapps.grades.tasks import compute_all_grades_for_course
 from openedx.core.djangoapps.credit.signals import on_course_publish
 from openedx.core.lib.gating import api as gating_api
 from util.module_utils import yield_dynamic_descriptor_descendants
+from .signals import GRADING_POLICY_CHANGED
 from xmodule.modulestore.django import SignalHandler, modulestore
 
+
 log = logging.getLogger(__name__)
 
 
@@ -83,3 +86,18 @@ def handle_item_deleted(**kwargs):
             gating_api.remove_prerequisite(module.location)
             # Remove any 'requires' course content milestone relationships
             gating_api.set_required_content(course_key, module.location, None, None)
+
+
+@receiver(GRADING_POLICY_CHANGED)
+def handle_grading_policy_changed(sender, **kwargs):
+    # pylint: disable=unused-argument
+    """
+    Receives signal and kicks off celery task to recalculate grades
+    """
+    course_key = kwargs.get('course_key')
+    result = compute_all_grades_for_course.apply_async(course_key=course_key)
+    log.info("Grades: Created {task_name}[{task_id}] with arguments {kwargs}".format(
+        task_name=compute_all_grades_for_course.name,
+        task_id=result.task_id,
+        kwargs=kwargs,
+    ))

cms/envs/common.py

@@ -1006,6 +1006,10 @@ INSTALLED_APPS = (
 
     # Unusual migrations
     'database_fixups',
+
+    # Customized celery tasks, including persisting failed tasks so they can
+    # be retried
+    'celery_utils',
 )
 
 
@@ -1302,3 +1306,8 @@ ENTERPRISE_API_CACHE_TIMEOUT = 3600  # Value is in seconds
 ############## Settings for the Discovery App ######################
 
 COURSE_CATALOG_API_URL = None
+
+############################# Persistent Grades ####################################
+
+# Queue to use for updating persistent grades
+RECALCULATE_GRADES_ROUTING_KEY = LOW_PRIORITY_QUEUE

lms/djangoapps/grades/management/commands/compute_grades.py

@@ -108,15 +108,11 @@ class Command(BaseCommand):
         all_args = []
         estimate_first_attempted = options['estimate_first_attempted']
         for course_key in self._get_course_keys(options):
-            enrollment_count = CourseEnrollment.objects.filter(course_id=course_key).count()
-            if enrollment_count == 0:
-                log.warning("No enrollments found for {}".format(course_key))
-            batch_size = self._latest_settings().batch_size if options.get('from_settings') else options['batch_size']
-            for offset in six.moves.range(options['start_index'], enrollment_count, batch_size):
-                # This is a tuple to reduce memory consumption.
-                # The dictionaries with their extra overhead will be created
-                # and consumed one at a time.
-                all_args.append((six.text_type(course_key), offset, batch_size))
+            # This is a tuple to reduce memory consumption.
+            # The dictionaries with their extra overhead will be created
+            # and consumed one at a time.
+            for task_arg_tuple in tasks._course_task_args(course_key, **options):
+                all_args.append(task_arg_tuple)
         all_args.sort(key=lambda x: hashlib.md5(b'{!r}'.format(x)))
         for args in all_args:
             yield {

lms/djangoapps/grades/tasks.py

@@ -10,12 +10,14 @@ from django.db.utils import DatabaseError
 from logging import getLogger
 
 log = getLogger(__name__)
+import six
 
 from celery_utils.logged_task import LoggedTask
 from celery_utils.persist_on_failure import PersistOnFailureTask
 from courseware.model_data import get_score
 from lms.djangoapps.course_blocks.api import get_course_blocks
 from lms.djangoapps.courseware import courses
+from lms.djangoapps.grades.config.models import ComputeGradesSetting
 from opaque_keys.edx.keys import CourseKey, UsageKey
 from opaque_keys.edx.locator import CourseLocator
 from openedx.core.djangoapps.monitoring_utils import (
@@ -54,6 +56,21 @@ class _BaseTask(PersistOnFailureTask, LoggedTask):  # pylint: disable=abstract-m
     abstract = True
 
 
+@task(base=_BaseTask)
+def compute_all_grades_for_course(**kwargs):
+    """
+    Compute grades for all students in the specified course.
+    Kicks off a series of compute_grades_for_course_v2 tasks
+    to cover all of the students in the course.
+    """
+    for course_key, offset, batch_size in _course_task_args(
+        course_key=kwargs.pop('course_key'),
+        kwargs=kwargs
+    ):
+        task_options = {'course_key': course_key, 'offset': offset, 'batch_size': batch_size}
+        compute_grades_for_course_v2.apply_async(kwargs=kwargs, **task_options)
+
+
 @task(base=_BaseTask, bind=True, default_retry_delay=30, max_retries=1)
 def compute_grades_for_course_v2(self, **kwargs):
     """
@@ -250,3 +267,21 @@ def _update_subsection_grades(course_key, scored_block_usage_key, only_if_higher
                     user=student,
                     subsection_grade=subsection_grade,
                 )
+
+
+def _course_task_args(course_key, **kwargs):
+    """
+    Helper function to generate course-grade task args.
+    """
+    from_settings = kwargs.pop('from_settings', True)
+    enrollment_count = CourseEnrollment.objects.filter(course_id=course_key).count()
+    if enrollment_count == 0:
+        log.warning("No enrollments found for {}".format(course_key))
+
+    if from_settings is False:
+        batch_size = kwargs.pop('batch_size', 100)
+    else:
+        batch_size = ComputeGradesSetting.current().batch_size
+
+    for offset in six.moves.range(0, enrollment_count, batch_size):
+        yield (six.text_type(course_key), offset, batch_size)

lms/djangoapps/grades/tests/test_tasks.py

@@ -31,9 +31,11 @@ from lms.djangoapps.grades.constants import ScoreDatabaseTableEnum
 from lms.djangoapps.grades.models import PersistentCourseGrade, PersistentSubsectionGrade
 from lms.djangoapps.grades.signals.signals import PROBLEM_WEIGHTED_SCORE_CHANGED
 from lms.djangoapps.grades.tasks import (
+    compute_all_grades_for_course,
     compute_grades_for_course_v2,
     recalculate_subsection_grade_v3,
-    RECALCULATE_GRADE_DELAY
+    RECALCULATE_GRADE_DELAY,
+    _course_task_args
 )
 
 
@@ -417,3 +419,23 @@ class ComputeGradesForCourseTest(HasCourseWithProblemsMixin, ModuleStoreTestCase
                     batch_size=batch_size,
                     offset=6,
                 )
+
+    @ddt.data(*xrange(1, 12, 3))
+    def test_compute_all_grades_for_course(self, batch_size):
+        self.set_up_course()
+        result = compute_all_grades_for_course.delay(
+            course_key=six.text_type(self.course.id),
+            batch_size=batch_size,
+        )
+        self.assertTrue(result.successful)
+
+    @ddt.data(*xrange(1, 12, 3))
+    def test_course_task_args(self, test_batch_size):
+        offset_expected = 0
+        for course_key, offset, batch_size in _course_task_args(
+            batch_size=test_batch_size, course_key=self.course.id, from_settings=False
+        ):
+            self.assertEqual(course_key, six.text_type(self.course.id))
+            self.assertEqual(batch_size, test_batch_size)
+            self.assertEqual(offset, offset_expected)
+            offset_expected += test_batch_size