Prevent compute_all_grades_for_course() task from being executed more than once every hour.

cms/djangoapps/contentstore/signals/handlers.py

 """ receivers of course_published and library_updated events in order to trigger indexing task """
 
-import logging
 from datetime import datetime
+from functools import wraps
+import logging
 
+from django.core.cache import cache
 from django.dispatch import receiver
 from pytz import UTC
 
@@ -19,6 +21,20 @@ from xmodule.modulestore.django import SignalHandler, modulestore
 
 log = logging.getLogger(__name__)
 
+GRADING_POLICY_COUNTDOWN_SECONDS = 3600
+
+
+def locked(expiry_seconds, key):
+    def task_decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            cache_key = '{}-{}'.format(func.__name__, kwargs[key])
+            if cache.add(cache_key, "true", expiry_seconds):
+                log.info('Locking task in cache with key: %s for %s seconds', cache_key, expiry_seconds)
+                return func(*args, **kwargs)
+        return wrapper
+    return task_decorator
+
 
 @receiver(SignalHandler.course_published)
 def listen_for_course_publish(sender, course_key, **kwargs):  # pylint: disable=unused-argument
@@ -90,6 +106,7 @@ def handle_item_deleted(**kwargs):
 
 
 @receiver(GRADING_POLICY_CHANGED)
+@locked(expiry_seconds=GRADING_POLICY_COUNTDOWN_SECONDS, key='course_key')
 def handle_grading_policy_changed(sender, **kwargs):
     # pylint: disable=unused-argument
     """
@@ -100,7 +117,7 @@ def handle_grading_policy_changed(sender, **kwargs):
         'event_transaction_id': unicode(get_event_transaction_id()),
         'event_transaction_type': unicode(get_event_transaction_type()),
     }
-    result = compute_all_grades_for_course.apply_async(kwargs=kwargs)
+    result = compute_all_grades_for_course.apply_async(kwargs=kwargs, countdown=GRADING_POLICY_COUNTDOWN_SECONDS)
     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,

cms/djangoapps/contentstore/tests/test_signals.py

+import ddt
+from mock import patch, Mock
+
+from cms.djangoapps.contentstore.signals.handlers import (
+    GRADING_POLICY_COUNTDOWN_SECONDS,
+    handle_grading_policy_changed
+)
+from student.models import CourseEnrollment, anonymous_id_for_user
+from student.tests.factories import UserFactory
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory
+
+
+@ddt.ddt
+class LockedTest(ModuleStoreTestCase):
+    def setUp(self):
+        super(LockedTest, self).setUp()
+        self.course = CourseFactory.create(
+            org='edx',
+            name='course',
+            run='run',
+        )
+        self.user = UserFactory.create()
+        CourseEnrollment.enroll(self.user, self.course.id)
+
+    @patch('cms.djangoapps.contentstore.signals.handlers.cache.add')
+    @patch('cms.djangoapps.contentstore.signals.handlers.cache.delete')
+    @patch('cms.djangoapps.contentstore.signals.handlers.compute_all_grades_for_course.apply_async')
+    @ddt.data(True, False)
+    def test_locked(self, lock_available, compute_grades_async_mock, delete_mock, add_mock):
+        add_mock.return_value = lock_available
+        sender = Mock()
+
+        handle_grading_policy_changed(sender, course_key=unicode(self.course.id))
+
+        cache_key = 'handle_grading_policy_changed-{}'.format(unicode(self.course.id))
+        self.assertEqual(lock_available, compute_grades_async_mock.called)
+        if lock_available:
+            add_mock.assert_called_once_with(cache_key, "true", GRADING_POLICY_COUNTDOWN_SECONDS)