Merge pull request #403 from mitodl/feature/skm/166_xanalytics

import xanalytics data

[delivers #96970016]

importer/api/__init__.py

@@ -17,6 +17,7 @@
 from lxml import etree
 from xbundle import XBundle, DESCRIPTOR_TAGS
 
+from importer.tasks import populate_xanalytics_fields
 from learningresources.api import (
     create_course,
     create_resource,
@@ -128,6 +129,7 @@ def import_course(bundle, repo_id, user_id, static_dir):
     )
     import_static_assets(course, static_dir)
     import_children(course, src, None, '')
+    populate_xanalytics_fields.delay(course.id)
     return course
 
 

importer/tasks.py

@@ -3,11 +3,63 @@
 """
 from __future__ import unicode_literals
 
+import json
+import logging
+
+from django.conf import settings
+import requests
+
 from lore.celery import async
-from importer.api import import_course_from_file
+from learningresources.api import update_xanalytics
+from xanalytics import send_request, get_result
+
+log = logging.getLogger(__name__)
+
+RETRY_LIMIT = 10
 
 
 @async.task
 def import_file(path, repo_id, user_id):
     """Asynchronously import a course."""
+    from importer.api import import_course_from_file
     import_course_from_file(path, repo_id, user_id)
+
+
+@async.task
+def populate_xanalytics_fields(course_id):
+    """
+    Initiate request to xanalytics API to get stats for a course,
+    then trigger async job to retrieve results when they become available.
+    Args:
+        course_id (int): primary key of a Course
+    """
+    if settings.XANALYTICS_URL != "":
+        token = send_request(settings.XANALYTICS_URL + "/create", course_id)
+        check_for_results.apply_async(
+            kwargs={"token": token, "wait": 5, "attempt": 1}, countdown=5)
+
+
+@async.task
+def check_for_results(token, wait, attempt):
+    """
+    Check for xanalytics results for a course.
+    Args:
+        token (string): Token received from xanalytics server.
+        wait (int): Seconds to wait before repository.
+        attempt (int): Attempt number, so we don't try forever.
+    """
+    resp = get_result(settings.XANALYTICS_URL + "/status", token)
+    if resp["status"] == "still busy" and attempt < RETRY_LIMIT:
+        attempt += 1
+        wait *= 2
+        check_for_results.apply_async(
+            kwargs={"token": token, "wait": wait, "attempt": attempt},
+            countdown=wait,
+        )
+    if resp["status"] == "complete":
+        content = requests.get(resp["url"]).content
+        try:
+            data = json.loads(content)
+            update_xanalytics(data)
+        except (ValueError, TypeError):
+            log.error("Unable to parse xanalytics response: %s", content)

importer/tests/test_xanalytics.py

@@ -0,0 +1,121 @@
+"""
+Tests for hitting the xanalytics Celery tasks.
+"""
+
+from __future__ import unicode_literals
+
+import json
+import logging
+
+from django.conf import settings
+import responses
+
+from importer.tasks import populate_xanalytics_fields, check_for_results
+from learningresources.tests.base import LoreTestCase
+
+log = logging.getLogger(__name__)
+# pylint: disable=no-self-use
+
+
+class TestAPI(LoreTestCase):
+    """
+    Tests using mocks to avoid actual HTTP requests.
+    """
+    def setUp(self):
+        """Override setting."""
+        super(TestAPI, self).setUp()
+        self.original_url = settings.XANALYTICS_URL
+        settings.XANALYTICS_URL = "http://example.com"
+
+    def tearDown(self):
+        """Restore original setting."""
+        super(TestAPI, self).tearDown()
+        settings.XANALYTICS_URL = self.original_url
+
+    @responses.activate
+    def test_populate_xanalytics_fields(self):
+        """Test kicking off a request."""
+        responses.add(
+            responses.POST,
+            'http://example.com/create',
+            body='{"token": "abcde"}',
+            content_type="application/json"
+        )
+        # We must mock the /status endpoint as well,
+        # because a status check is queued when /create is called.
+        reply = {"status": "still busy"}
+        responses.add(
+            responses.POST,
+            'http://example.com/status',
+            body=json.dumps(reply),
+            content_type="application/json"
+        )
+        populate_xanalytics_fields(1234)
+
+    @responses.activate
+    def test_check_for_results_waiting(self):
+        """Test kicking off a request."""
+        reply = {"status": "still busy"}
+        responses.add(
+            responses.POST,
+            'http://example.com/status',
+            body=json.dumps(reply),
+            content_type="application/json"
+        )
+        check_for_results("abcde", 10, 1)
+
+    @responses.activate
+    def test_check_for_results_done(self):
+        """Test kicking off a request."""
+        file_reply = {
+            "course_id": "123",
+            "module_medata": [
+                {
+                    "module_id": "1",
+                    "xa_nr_views": "3",
+                    "xa_nr_attempts": "25"
+                },
+                {
+                    "module_id": "2",
+                    "xa_nr_views": "7",
+                    "xa_nr_attempts": "99"
+
+                }
+            ]
+        }
+        responses.add(
+            responses.GET,
+            'http://example.com/foo.json',
+            body=json.dumps(file_reply),
+            content_type="application/json"
+        )
+
+        reply = {"status": "complete", "url": "http://example.com/foo.json"}
+        responses.add(
+            responses.POST,
+            'http://example.com/status',
+            body=json.dumps(reply),
+            content_type="application/json"
+        )
+        check_for_results("abcde", 10, 1)
+
+    @responses.activate
+    def test_check_for_results_bad(self):
+        """Test kicking off a request."""
+        responses.add(
+            responses.GET,
+            'http://example.com/foo.json',
+            body='{"invalid": "JSON", "because": "developer comma",}',
+            content_type="application/json"
+        )
+        reply = {"status": "complete", "url": "http://example.com/foo.json"}
+        responses.add(
+            responses.POST,
+            'http://example.com/status',
+            body=json.dumps(reply),
+            content_type="application/json"
+        )
+        self.assertRaises(
+            ValueError,
+            check_for_results("abcde", 10, 1)
+        )

learningresources/api.py

@@ -307,6 +307,29 @@ def import_static_assets(course, path):
                 create_static_asset(course.id, django_file)
 
 
+def update_xanalytics(data):
+    """
+    Update xanalytics fields for a LearningResource.
+    Args:
+        data (dict): dict from JSON file from xanalytics
+    Returns:
+        count (int): number of records updated
+    """
+    vals = data.get("module_medata", [])
+    course_number = data.get("course_id", "")
+    count = 0
+    for rec in vals:
+        resource_key = rec.pop("module_id")
+        count = LearningResource.objects.filter(
+            uuid=resource_key,
+            course__course_number=course_number,
+
+        ).update(**rec)
+        if count is None:
+            count = 0
+    return count
+
+
 def join_description_paths(*args):
     """
     Helper function to format the description path.

learningresources/tests/test_xanalytics.py

@@ -0,0 +1,59 @@
+"""
+Tests for utilities within the API.
+"""
+
+import logging
+
+from learningresources.tests.base import LoreTestCase
+
+from learningresources.api import update_xanalytics
+from learningresources.models import LearningResource
+
+log = logging.getLogger(__name__)
+
+
+class TestXanalyticsData(LoreTestCase):
+    """
+    Test good and bad data.
+    """
+    def test_good(self):
+        """Reasonable data."""
+        self.resource.uuid = "1"
+        self.resource.save()
+        resource = LearningResource.objects.get(id=self.resource.id)
+
+        self.assertEqual(resource.xa_nr_views, 0)
+        self.assertEqual(resource.xa_nr_attempts, 0)
+
+        reasonable = {
+            "course_id": self.course.course_number,
+            "module_medata": [
+                {
+                    "module_id": "1",
+                    "xa_nr_views": "3",
+                    "xa_nr_attempts": "25"
+                },
+            ]
+        }
+
+        self.assertEqual(update_xanalytics(reasonable), 1)
+        resource = LearningResource.objects.get(id=self.resource.id)
+        self.assertEqual(resource.xa_nr_views, 3)
+        self.assertEqual(resource.xa_nr_attempts, 25)
+
+    def test_empty(self):
+        """Empty dict."""
+        self.assertEqual(update_xanalytics({}), 0)
+
+    def test_course_id(self):
+        """Missing/bad course ID."""
+        data = {
+            "module_medata": [
+                {
+                    "module_id": "1",
+                    "xa_nr_views": "3",
+                    "xa_nr_attempts": "25"
+                },
+            ]
+        }
+        self.assertEqual(update_xanalytics(data), 0)

lore/settings.py

@@ -354,3 +354,4 @@ def get_var(name, default):
 }
 HAYSTACK_SIGNAL_PROCESSOR = 'haystack.signals.RealtimeSignalProcessor'
 ALLOW_CACHING = get_var("ALLOW_CACHING", get_var("ALLOW_CACHING", False))
+XANALYTICS_URL = get_var('XANALYTICS_URL', "")

requirements.txt

@@ -13,6 +13,7 @@ git+https://github.com/bpeschier/django-compressor-requirejs@889d5edc4f2acaa961c
 psycopg2==2.6.1
 PyYAML==3.11
 redis==2.10.3
+requests==2.7.0
 static3==0.6.1
 tox>=2.0.2,<3.0.0
 uwsgi==2.0.11

test_requirements.txt

@@ -17,3 +17,4 @@ ipdb
 ipython
 urltools
 semantic_version
+responses

xanalytics/__init__.py

@@ -0,0 +1,58 @@
+"""
+Functions for communicating with the xanalytics API.
+"""
+import logging
+import os
+from tempfile import mkstemp
+
+from django.conf import settings
+from requests.exceptions import ConnectionError as RequestsConnectionError
+import requests
+
+
+log = logging.getLogger(__name__)
+
+
+def _call(url, data):
+    """
+    Make call via requests, trapping common errors.
+    Args:
+        url (unicode): URL
+        data (dict): POST parameters
+    Returns:
+        result (dict): Results read as JSON
+    """
+    try:
+        resp = requests.post(url=url, data=data)
+        try:
+            return resp.json()
+        except ValueError as ex:
+            log.error("Bad JSON returned: %s", ex.args)
+    except RequestsConnectionError as ex:
+        log.error("Unable to connect to xanalytics server: %s", ex.args)
+
+    # Fallback in case things fail.
+    return {}
+
+
+def send_request(url, course_id):
+    """
+    Send initial request to xanalytics API.
+    Args:
+        url (string): URL of xanalytics API
+        course_id (int): Course primary key.
+    Returns:
+        token (unicode): Job token from xanalytics.
+    """
+    return _call(url=url, data={"course_id": course_id})
+
+
+def get_result(url, token):
+    """
+    Get result from xanalytics for token.
+    Args:
+        token (unicode): Token from xanalytics job.
+    Returns:
+        data (dict): Dictionary from JSON object provided by xanalytics.
+    """
+    return _call(url=url, data={"token": token})