Use instructor_tasks for canvas work (#183)

* Add per_page to paginated requests

* Static method

* Fix email lowercase mismatch

* Use instructor_tasks to handle canvas work

* Fix transaction error

* Decorator needs to be at the top

* Attempt to fix polling of tasks

* More task bug fixes

* Fix typo

* Fix course key bug

* Update message for push edx grades command

* Use course_key so sync_canvas_enrollments uses the same message

* Format the submitted time

Author: George Schneeloch

lms/djangoapps/canvas_integration/api.py

@@ -150,7 +150,7 @@ def push_edx_grades_to_canvas(course):
     for subsection_block, user_grade_dict in subsection_block_user_grades.items():
         grade_update_payloads[subsection_block] = dict(
             update_grade_payload_kv(
-                enrolled_user_dict[student_user.email],
+                enrolled_user_dict[student_user.email.lower()],
                 grade.percent_graded
             )
             for student_user, grade in user_grade_dict.items()

lms/djangoapps/canvas_integration/client.py

@@ -1,7 +1,6 @@
 import logging
-import time
 import pytz
-from urllib.parse import urljoin
+from urllib.parse import urlencode, urljoin, urlparse, parse_qs
 import requests
 
 from django.conf import settings
@@ -28,10 +27,31 @@ def get_canvas_session():
         })
         return session
 
+    @staticmethod
+    def _add_per_page(url, per_page):
+        """
+        Add per_page query parameter to override default value of 10
+
+        Args:
+            url (str): The url to update
+            per_page (int): The new per_page value
+
+        Returns:
+            str: The updated URL
+        """
+        pieces = urlparse(url)
+        query = parse_qs(pieces.query)
+        query['per_page'] = per_page
+        query_string = urlencode(query, doseq=True)
+        pieces = pieces._replace(query=query_string)
+        return pieces.geturl()
+
     def _paginate(self, url, *args, **kwargs):
         """
         Iterate over the paginated results of a request
         """
+        url = self._add_per_page(url, 100)  # increase per_page to 100 from default of 10
+
         items = []
         while url:
             resp = self.session.get(url, *args, **kwargs)
@@ -42,10 +62,6 @@ def _paginate(self, url, *args, **kwargs):
             for link in links:
                 if link["rel"] == "next":
                     url = link["url"]
-                    # TODO: delay at all? Canvas docs don't say it's necessary, just that the throttled amount is
-                    # respected. If we go over it should trigger a 403 which will raise an exception here.
-                    # 0.2 seconds per request is just a guess to space out requests on the Canvas API
-                    time.sleep(0.2)
 
         return items
 

lms/djangoapps/canvas_integration/constants.py

@@ -1 +1,7 @@
 DEFAULT_ASSIGNMENT_POINTS = 1
+TASK_TYPE_SYNC_CANVAS_ENROLLMENTS = "sync_canvas_enrollments"
+TASK_TYPE_PUSH_EDX_GRADES_TO_CANVAS = "push_edx_grades_to_canvas"
+CANVAS_TASK_TYPES = [
+    TASK_TYPE_SYNC_CANVAS_ENROLLMENTS,
+    TASK_TYPE_PUSH_EDX_GRADES_TO_CANVAS,
+]

lms/djangoapps/canvas_integration/task_helpers.py

@@ -0,0 +1,60 @@
+"""
+Helper functions for canvas integration tasks
+"""
+from time import time
+
+from canvas_integration import api
+from courseware.courses import get_course_by_id
+from instructor_task.tasks_helper.runner import TaskProgress
+
+
+def sync_canvas_enrollments(_xmodule_instance_args, _entry_id, course_id, task_input, action_name):
+    """Partial function to sync canvas enrollments"""
+    start_time = time()
+    num_reports = 1
+    task_progress = TaskProgress(action_name, num_reports, start_time)
+    api.sync_canvas_enrollments(
+        course_key=task_input["course_key"],
+        canvas_course_id=task_input["canvas_course_id"],
+        unenroll_current=task_input["unenroll_current"],
+    )
+    # for simplicity, only one task update for now when everything is done
+    return task_progress.update_task_state(extra_meta={
+        "step": "Done"
+    })
+
+
+def push_edx_grades_to_canvas(_xmodule_instance_args, _entry_id, course_id, task_input, action_name):
+    """Partial function to push edX grades to canvas"""
+    start_time = time()
+    num_reports = 1
+    task_progress = TaskProgress(action_name, num_reports, start_time)
+    course = get_course_by_id(course_id)
+    assignment_grades_updated, created_assignments = api.push_edx_grades_to_canvas(
+        course=course
+    )
+    results = {}
+    if assignment_grades_updated:
+        grade_update_results = {}
+        for subsection_block, grade_update_response in assignment_grades_updated.items():
+            if grade_update_response.ok:
+                message = "updated"
+            else:
+                message = {"error": grade_update_response.status_code}
+            grade_update_results[subsection_block.display_name] = message
+        results["grades"] = grade_update_results
+    if created_assignments:
+        created_assignment_results = {}
+        for subsection_block, new_assignment_response in created_assignments.items():
+            if new_assignment_response.ok:
+                message = "created"
+            else:
+                message = {"error": new_assignment_response.status_code}
+            created_assignment_results[subsection_block.display_name] = message
+        results["assignments"] = created_assignment_results
+
+    # for simplicity, only one task update for now when everything is done
+    return task_progress.update_task_state(extra_meta={
+        "step": "Done",
+        "results": results
+    })

lms/djangoapps/canvas_integration/tasks.py

@@ -1,30 +1,74 @@
 """Tasks for canvas"""
-
-import logging
 import hashlib
+import logging
 from functools import partial
 
 from celery import task
-from django.conf import settings
-from django.utils.translation import ugettext_noop
+from opaque_keys.edx.locator import CourseLocator
+
+from canvas_integration import task_helpers
+from canvas_integration.constants import (
+    TASK_TYPE_SYNC_CANVAS_ENROLLMENTS,
+    TASK_TYPE_PUSH_EDX_GRADES_TO_CANVAS,
+)
+from courseware.courses import get_course_by_id
+from instructor_task.api_helper import submit_task
+from instructor_task.tasks_base import BaseInstructorTask
+from instructor_task.tasks_helper.runner import run_main_task
 
-from canvas_integration import api
 
 TASK_LOG = logging.getLogger('edx.celery.task')
 
 
-@task
-def sync_canvas_enrollments(course_key, canvas_course_id, unenroll_current):
+def run_sync_canvas_enrollments(request, course_key, canvas_course_id, unenroll_current):
+    """
+    Submit a task to start syncing canvas enrollments
+    """
+    task_type = TASK_TYPE_SYNC_CANVAS_ENROLLMENTS
+    task_class = sync_canvas_enrollments_task
+    task_input = {
+        "course_key": course_key,
+        "canvas_course_id": canvas_course_id,
+        "unenroll_current": unenroll_current
+    }
+    task_key = hashlib.md5(course_key.encode("utf8")).hexdigest()
+    TASK_LOG.debug("Submitting task to sync canvas enrollments")
+    return submit_task(request, task_type, task_class, course_key, task_input, task_key)
+
+
+@task(base=BaseInstructorTask)
+def sync_canvas_enrollments_task(entry_id, xmodule_instance_args):
     """
     Fetch enrollments from canvas and update
+    """
+    action_name = "sync_canvas_enrollments"
+    TASK_LOG.info("Running task to sync Canvas enrollments")
+    task_fn = partial(task_helpers.sync_canvas_enrollments, xmodule_instance_args)
+    return run_main_task(entry_id, task_fn, action_name)
+
 
-    Args:
-        course_key (str): The edX course key
-        canvas_course_id (int): The canvas course id
-        unenroll_current (bool): If true, unenroll existing students TODO
+def run_push_edx_grades_to_canvas(request, course_id):
+    """
+    Submit a task to start pushing edX grades to Canvas
+    """
+    task_type = TASK_TYPE_PUSH_EDX_GRADES_TO_CANVAS
+    task_class = push_edx_grades_to_canvas_task
+    task_input = {
+        # course_key is already passed into the task, but we need to put it in task_input as well
+        # so the instructor task status can be properly calculated instead of being marked incomplete
+        "course_key": str(course_id)
+    }
+    task_key = hashlib.md5(course_id.encode("utf8")).hexdigest()
+    TASK_LOG.debug("Submitting task to push edX grades to Canvas")
+    return submit_task(request, task_type, task_class, course_id, task_input, task_key)
+
+
+@task(base=BaseInstructorTask)
+def push_edx_grades_to_canvas_task(entry_id, xmodule_instance_args):
+    """
+    Push edX grades to Canvas
     """
-    api.sync_canvas_enrollments(
-        course_key=course_key,
-        canvas_course_id=canvas_course_id,
-        unenroll_current=unenroll_current,
-    )
+    action_name = "push_edx_grades_to_canvas"
+    TASK_LOG.info("Running task to push edX grades to Canvas")
+    task_fn = partial(task_helpers.push_edx_grades_to_canvas, xmodule_instance_args)
+    return run_main_task(entry_id, task_fn, action_name)

lms/djangoapps/canvas_integration/views.py

@@ -1,18 +1,27 @@
+"""Views for canvas integration"""
+import logging
+
 from django.contrib.auth.models import User
+from django.db import transaction
+from django.utils.translation import ugettext as _
 from django.views.decorators.cache import cache_control
 from django.views.decorators.csrf import ensure_csrf_cookie
 from django.views.decorators.http import require_POST
 from opaque_keys.edx.locator import CourseLocator
 
-from canvas_integration.client import CanvasClient
+from lms.djangoapps.canvas_integration.client import CanvasClient
 from courseware.courses import get_course_by_id
-from canvas_integration import api
+from lms.djangoapps.canvas_integration import tasks
+from lms.djangoapps.instructor_task.api_helper import AlreadyRunningError
 from lms.djangoapps.instructor.views.api import require_course_permission
 from lms.djangoapps.instructor import permissions
 from student.models import CourseEnrollment, CourseEnrollmentAllowed
 from util.json_request import JsonResponse
 
 
+log = logging.getLogger(__name__)
+
+
 def _get_edx_enrollment_data(email, course_key):
     """Helper function to look up some info regarding whether a user with a email address is enrolled in edx"""
     user = User.objects.filter(email=email).first()
@@ -35,7 +44,6 @@ def list_canvas_enrollments(request, course_id):
     course_key = CourseLocator.from_string(course_id)
     course = get_course_by_id(course_key)
     if not course.canvas_course_id:
-        # TODO: better exception class?
         raise Exception("No canvas_course_id set for course {}".format(course_id))
 
     client = CanvasClient(canvas_course_id=course.canvas_course_id)
@@ -48,6 +56,7 @@ def list_canvas_enrollments(request, course_id):
     return JsonResponse(results)
 
 
+@transaction.non_atomic_requests
 @require_POST
 @ensure_csrf_cookie
 @cache_control(no_cache=True, no_store=True, must_revalidate=True)
@@ -60,14 +69,21 @@ def add_canvas_enrollments(request, course_id):
     course_key = CourseLocator.from_string(course_id)
     course = get_course_by_id(course_key)
     if not course.canvas_course_id:
-        # TODO: better exception class?
         raise Exception("No canvas_course_id set for course {}".format(course_id))
-    api.sync_canvas_enrollments(
-        course_key=course_id,
-        canvas_course_id=course.canvas_course_id,
-        unenroll_current=unenroll_current,
-    )  # WARNING: this will block the web thread
-    return JsonResponse({"status": "success"})
+
+    try:
+        tasks.run_sync_canvas_enrollments(
+            request=request,
+            course_key=course_id,
+            canvas_course_id=course.canvas_course_id,
+            unenroll_current=unenroll_current,
+        )
+        log.info("Syncing canvas enrollments for course %s", course_id)
+        success_status = _("Syncing canvas enrollments")
+        return JsonResponse({"status": success_status})
+    except AlreadyRunningError:
+        already_running_status = _("Syncing canvas enrollments. See Pending Tasks below to view the status.")
+        return JsonResponse({"status": already_running_status})
 
 
 @ensure_csrf_cookie
@@ -79,7 +95,6 @@ def list_canvas_assignments(request, course_id):
     course = get_course_by_id(course_key)
     client = CanvasClient(canvas_course_id=course.canvas_course_id)
     if not course.canvas_course_id:
-        # TODO: better exception class?
         raise Exception("No canvas_course_id set for course {}".format(course_id))
     return JsonResponse(client.list_canvas_assignments())
 
@@ -94,11 +109,12 @@ def list_canvas_grades(request, course_id):
     course = get_course_by_id(course_key)
     client = CanvasClient(canvas_course_id=course.canvas_course_id)
     if not course.canvas_course_id:
-        # TODO: better exception class?
         raise Exception("No canvas_course_id set for course {}".format(course_id))
     return JsonResponse(client.list_canvas_grades(assignment_id=assignment_id))
 
 
+@transaction.non_atomic_requests
+@require_POST
 @ensure_csrf_cookie
 @cache_control(no_cache=True, no_store=True, must_revalidate=True)
 @require_course_permission(permissions.OVERRIDE_GRADES)
@@ -107,29 +123,15 @@ def push_edx_grades(request, course_id):
     course_key = CourseLocator.from_string(course_id)
     course = get_course_by_id(course_key)
     if not course.canvas_course_id:
-        # TODO: better exception class?
         raise Exception("No canvas_course_id set for course {}".format(course_id))
-    assignment_grades_updated, created_assignments = api.push_edx_grades_to_canvas(
-        course=course
-    )
-
-    results = {}
-    if assignment_grades_updated:
-        grade_update_results = {}
-        for subsection_block, grade_update_response in assignment_grades_updated.items():
-            if grade_update_response.ok:
-                message = "updated"
-            else:
-                message = {"error": grade_update_response.status_code}
-            grade_update_results[subsection_block.display_name] = message
-        results["grades"] = grade_update_results
-    if created_assignments:
-        created_assignment_results = {}
-        for subsection_block, new_assignment_response in created_assignments.items():
-            if new_assignment_response.ok:
-                message = "created"
-            else:
-                message = {"error": new_assignment_response.status_code}
-            created_assignment_results[subsection_block.display_name] = message
-        results["assignments"] = created_assignment_results
-    return JsonResponse(results)
+    try:
+        tasks.run_push_edx_grades_to_canvas(
+            request=request,
+            course_id=course_id,
+        )
+        log.info("Pushing edX grades to canvas for course %s", course_id)
+        success_status = _("Pushing edX grades to canvas")
+        return JsonResponse({"status": success_status})
+    except AlreadyRunningError:
+        already_running_status = _("Pushing edX grades to canvas. See Pending Tasks below to view the status.")
+        return JsonResponse({"status": already_running_status})

lms/djangoapps/instructor/views/api.py

@@ -1912,6 +1912,8 @@ def list_instructor_tasks(request, course_id):
         - `problem_location_str` and `unique_student_identifier` lists task
             history for problem AND student (intersection)
     """
+    include_remote_gradebook = request.GET.get('include_remote_gradebook') is not None
+    include_canvas = request.GET.get('include_canvas') is not None
     course_id = CourseKey.from_string(course_id)
     problem_location_str = strip_if_string(request.POST.get('problem_location_str', False))
     student = request.POST.get('unique_student_identifier', None)
@@ -1934,6 +1936,16 @@ def list_instructor_tasks(request, course_id):
         else:
             # Specifying for single problem's history
             tasks = task_api.get_instructor_task_history(course_id, module_state_key)
+    elif include_remote_gradebook:
+        tasks = task_api.get_running_instructor_rgb_tasks(
+            course_id,
+            user=request.user
+        )
+    elif include_canvas:
+        tasks = task_api.get_running_instructor_canvas_tasks(
+            course_id,
+            user=request.user
+        )
     else:
         # If no problem or student, just get currently running tasks
         tasks = task_api.get_running_instructor_tasks(course_id)

lms/djangoapps/instructor/views/instructor_dashboard.py

@@ -710,6 +710,10 @@ def _section_canvas_integration(course):
         "list_canvas_enrollments_url": reverse("list_canvas_enrollments", kwargs={"course_id": course.id}),
         "list_canvas_assignments_url": reverse("list_canvas_assignments", kwargs={"course_id": course.id}),
         "list_canvas_grades_url": reverse("list_canvas_grades", kwargs={"course_id": course.id}),
+        'list_instructor_tasks_url': '{}?include_canvas=true'.format(reverse(
+            'list_instructor_tasks',
+            kwargs={'course_id': course.id}
+        )),
         "push_edx_grades_url": reverse(
             "push_edx_grades", kwargs={"course_id": course.id}
         ),