Merge pull request #802 from mitodl/refactor/gs/tasks_api

Added tasks API

Author: George Schneeloch

apiary.apib

@@ -89,6 +89,86 @@ Page numbering is 1-based. Omitting the `?page` parameter will return the first
             ]
         }
 
+## Group Tasks
+
+## Tasks Collection [/tasks/]
+
+### List All Tasks [GET]
+
+List all recent tasks for a user.
+
++ Response 200 (application/json)
+
+        {
+            "count": 1,
+            "next": null,
+            "previous": null,
+            "results": [
+                {
+                    "id": "45e3c830-0ff8-4d84-85b4-c0a6e3ce81b3",
+                    "status": "processing",
+                    "result": {
+                        "url": "/media/resource_exports/sarah_exports.tar.gz",
+                        "collision": false
+                    },
+                    "task_type": "resource_export",
+                    "task_info": {
+                        "repo_slug": "repo",
+                        "ids": [
+                            23517,
+                            23518
+                        ]
+                    }
+                }
+            ]
+        }
+
+### Create a New Task [POST]
+
+Queue a new task.
+
++ Request (application/json)
+
+        {
+            "task_info": {
+                "repo_slug": "repo",
+                "ids": [1]
+            },
+            "task_type": "resource_export"
+        }
+
++ Response 200 (application/json)
+
+        {
+            "id": "45e3c830-0ff8-4d84-85b4-c0a6e3ce81b3"
+        }
+
+## Task [/tasks/{task_id}/]
+
++ Parameters
+    + task_id: `45e3c830-0ff8-4d84-85b4-c0a6e3ce81b3` (string, required) - task identifier
+
+### Retrieve a Task [GET]
+
++ Response 200 (application/json)
+
+        {
+            "id": "45e3c830-0ff8-4d84-85b4-c0a6e3ce81b3",
+            "status": "processing",
+            "result": {
+                "url": "/media/resource_exports/sarah_exports.tar.gz",
+                "collision": false
+            },
+            "task_type": "resource_export",
+            "task_info": {
+                "repo_slug": "repo",
+                "ids": [
+                    23517,
+                    23518
+                ]
+            }
+        }
+
 ## Group Repositories
 
 ## Repository Collection [/repositories/]
@@ -477,6 +557,8 @@ Clears the export list for this repository.
 
 ## LearningResourceExportTasks Collection [/repositories/{repo_slug}/learning_resource_export_tasks/]
 
+Deprecated. Use /api/v1/tasks/ instead.
+
 + Parameters
     + repo_slug: `physics-1` (string, required) - slug for the repository
 
@@ -515,7 +597,9 @@ Queue a new LearningResourceExportTask task for the given LearningResource ids.
             "id": "45e3c830-0ff8-4d84-85b4-c0a6e3ce81b3"
         }
 
-## LearningResourceExportTask [/repositories/{repo_slug}/learning_resource_export_tasks/{task_id}/
+## LearningResourceExportTask [/repositories/{repo_slug}/learning_resource_export_tasks/{task_id}/]
+
+Deprecated. Use /api/v1/tasks/ instead.
 
 + Parameters
     + repo_slug: `physics-1` (string, required) - slug for the repository

exporter/tasks.py

@@ -4,6 +4,8 @@
 
 from __future__ import unicode_literals
 
+from django.core.files.storage import default_storage
+
 from lore.celery import async
 from exporter.api import export_resources_to_tarball
 
@@ -18,8 +20,14 @@ def export_resources(learning_resources, username):
             LearningResources to export in tarball
         username (unicode): Name of user
     Returns:
-        (unicode, bool):
-            First item is newly created temp directory with files inside of it.
-            Second item is True if a static asset collision was detected.
+        dict:
+            name is path of tarball.
+            url is URL of tarball using django-storage.
+            collision is True if a static asset collision was detected.
     """
-    return export_resources_to_tarball(learning_resources, username)
+    name, collision = export_resources_to_tarball(learning_resources, username)
+    return {
+        "name": name,
+        "url": default_storage.url(name),
+        "collision": collision
+    }

exporter/tests/test_export.py

@@ -175,8 +175,10 @@ def test_export_task(self):
         """Test exporting resources task."""
         resources = LearningResource.objects.all()
 
-        path, collision = export_resources.delay(
+        result = export_resources.delay(
             resources, self.user.username).get()
+        path = result['name']
+        collision = result['collision']
         tempdir = mkdtemp()
 
         self.assertTrue(collision)

rest/serializers.py

@@ -21,6 +21,8 @@
     SerializerMethodField,
     IntegerField,
     FloatField,
+    DictField,
+    BooleanField,
 )
 
 from rest.util import LambdaDefault, RequiredBooleanField
@@ -308,6 +310,16 @@ class LearningResourceExportTaskSerializer(Serializer):
     id = CharField()
     status = CharField()
     url = CharField()
+    collision = BooleanField()
+
+
+class TaskSerializer(Serializer):
+    """Serializer for tasks."""
+    id = CharField()
+    status = CharField()
+    result = DictField()
+    task_type = CharField()
+    task_info = DictField()
 
 
 class RepositorySearchSerializer(Serializer):

rest/tasks.py

@@ -0,0 +1,211 @@
+"""
+Functions to manipulate tasks via REST API.
+"""
+
+from __future__ import unicode_literals
+
+from celery.result import AsyncResult
+from celery.states import FAILURE, SUCCESS, REVOKED
+from django.contrib.auth.models import User
+from django.http.response import Http404
+from rest_framework.exceptions import ValidationError
+
+from exporter.tasks import export_resources
+from learningresources.api import get_repo
+from learningresources.models import LearningResource
+
+TASK_KEY = 'tasks'
+EXPORT_TASK_TYPE = 'resource_export'
+EXPORTS_KEY = 'learning_resource_exports'
+
+
+def create_initial_task_dict(task, task_type, task_info):
+    """
+    Create initial task data about a newly created Celery task.
+
+    Args:
+        task (Task): A Celery task.
+        task_type (unicode): Type of task.
+        task_info (dict): Extra information about a task.
+    Returns:
+        dict: Initial data about task.
+    """
+
+    result = None
+    if task.successful():
+        result = task.get()
+    elif task.failed():
+        result = {'error': str(task.result)}
+
+    return {
+        "id": task.id,
+        "initial_state": task.state,
+        "task_type": task_type,
+        "task_info": task_info,
+        "result": result
+    }
+
+
+def create_task_result_dict(initial_data):
+    """
+    Convert initial data we put in session to dict for REST API.
+    This will use the id to look up current data about task to return
+    to user.
+
+    Args:
+        task (dict): Initial data about task stored in session.
+    Returns:
+        dict: Updated data about task.
+    """
+    initial_state = initial_data['initial_state']
+    task_id = initial_data['id']
+    task_type = initial_data['task_type']
+    task_info = initial_data['task_info']
+
+    state = "processing"
+    result = None
+    # initial_state is a workaround for EagerResult used in testing.
+    # In production initial_state should usually be pending.
+    async_result = AsyncResult(task_id)
+
+    if initial_state == SUCCESS:
+        state = "success"
+        result = initial_data['result']
+    elif initial_state in (FAILURE, REVOKED):
+        state = "failure"
+        result = initial_data['result']
+    elif async_result.successful():
+        state = "success"
+        result = async_result.get()
+    elif async_result.failed():
+        state = "failure"
+        result = {'error': str(async_result.result)}
+
+    return {
+        "id": task_id,
+        "status": state,
+        "result": result,
+        "task_type": task_type,
+        "task_info": task_info
+    }
+
+
+def get_tasks(session):
+    """
+    Get initial task data for session.
+
+    Args:
+        session (SessionStore): The request session.
+    Returns:
+        dict:
+            The initial task data stored in session for all user's tasks. The
+            keys are task ids and the values are initial task data.
+    """
+    try:
+        return session[TASK_KEY]
+    except KeyError:
+        return {}
+
+
+def get_task(session, task_id):
+    """
+    Get initial task data for a single task.
+
+    Args:
+        session (SessionStore): The request session.
+        task_id (unicode): The task id.
+    Returns:
+        dict: The initial task data stored in session.
+    """
+    try:
+        return session[TASK_KEY][task_id]
+    except KeyError:
+        return None
+
+
+def track_task(session, task, task_type, task_info):
+    """
+    Add a Celery task to the session.
+
+    Args:
+        session (SessionStore): The request session.
+        task_type (unicode): The type of task being started.
+        task_info (dict): Extra information about the task.
+    Returns:
+        dict: The initial task data (will also be stored in session).
+    """
+    initial_data = create_initial_task_dict(task, task_type, task_info)
+    if TASK_KEY not in session:
+        session[TASK_KEY] = {}
+    session[TASK_KEY][task.id] = initial_data
+    session.modified = True
+    return initial_data
+
+
+def create_task(session, user_id, task_type, task_info):
+    """
+    Start a new Celery task from REST API.
+
+    Args:
+        session (SessionStore): The request session.
+        user_id (int): The id for user creating task.
+        task_type (unicode): The type of task being started.
+        task_info (dict): Extra information about the task.
+    Returns:
+        dict: The initial task data (will also be stored in session).
+    """
+
+    if task_type == EXPORT_TASK_TYPE:
+        try:
+            repo_slug = task_info['repo_slug']
+        except KeyError:
+            raise ValidationError("Missing repo_slug")
+
+        # Verify repository ownership.
+        get_repo(repo_slug, user_id)
+
+        try:
+            exports = set(session[EXPORTS_KEY][repo_slug])
+        except KeyError:
+            exports = set()
+
+        try:
+            ids = task_info['ids']
+        except KeyError:
+            raise ValidationError("Missing ids")
+
+        for resource_id in ids:
+            if resource_id not in exports:
+                raise ValidationError("id {id} is not in export list".format(
+                    id=resource_id
+                ))
+
+        learning_resources = LearningResource.objects.filter(id__in=ids).all()
+        user = User.objects.get(id=user_id)
+        result = export_resources.delay(learning_resources, user.username)
+
+        # Put new task in session.
+        initial_data = track_task(session, result, task_type, task_info)
+
+        return initial_data
+    else:
+        raise ValidationError("Unknown task_type {task_type}".format(
+            task_type=task_type
+        ))
+
+
+def remove_task(session, task_id):
+    """
+    Cancel task and remove task from task list.
+
+    Args:
+        session (SessionStore): The request session.
+        task_id (int): The task id.
+    """
+    tasks = session.get(TASK_KEY, {})
+    if task_id not in tasks:
+        raise Http404
+
+    AsyncResult(task_id).revoke()
+    del tasks[task_id]
+    session[TASK_KEY] = tasks