Merge aacf4e0 into d9d7f83

qiita_core/testing.py

@@ -39,12 +39,24 @@ def wait_for_prep_information_job(prep_id, raise_if_none=True):
         payload = loads(res)
         job_id = payload['job_id']
         if payload['is_qiita_job']:
-            job = ProcessingJob(job_id)
-            while job.status not in ('success', 'error'):
-                sleep(0.05)
+            wait_for_processing_job(job_id)
         else:
             redis_info = loads(r_client.get(job_id))
             while redis_info['status_msg'] == 'Running':
                 sleep(0.05)
                 redis_info = loads(r_client.get(job_id))
         sleep(0.05)
+
+
+def wait_for_processing_job(job_id):
+    """Waits until a processing job is completed
+
+    Parameters
+    ----------
+    job_id : str
+        Job id
+    """
+    job = ProcessingJob(job_id)
+    while job.status not in ('success', 'error'):
+        sleep(0.05)
+    sleep(0.05)

qiita_db/handlers/processing_job.py

@@ -7,9 +7,16 @@
 # -----------------------------------------------------------------------------
 
 from json import loads
+from sys import exc_info
+import traceback
 
 from tornado.web import HTTPError
 
+# We agreed before that qiita db should never import from
+# qiita_ware. However, this is part of the rest API and I think it
+# is acceptable to import from qiita_ware, specially to offload
+# processing to the ipython cluster
+from qiita_ware.context import safe_submit
 import qiita_db as qdb
 from .oauth2 import OauthBaseHandler, authenticate_oauth
 
@@ -45,6 +52,30 @@ def _get_job(job_id):
     return job
 
 
+def _job_completer(job_id, payload):
+    """Completes a job
+
+    Parameters
+    ----------
+    job_id : str
+        The job to complete
+    payload : dict
+        The parameters of the HTTP POST request that is completing the job
+    """
+    payload_success = payload['success']
+    if payload_success:
+        artifacts = payload['artifacts']
+        error = None
+    else:
+        artifacts = None
+        error = payload['error']
+    job = qdb.processing_job.ProcessingJob(job_id)
+    try:
+        job.complete(payload_success, artifacts, error)
+    except:
+        job._set_error(traceback.format_exception(*exc_info()))
+
+
 class JobHandler(OauthBaseHandler):
     @authenticate_oauth
     def get(self, job_id):
@@ -133,18 +164,12 @@ def post(self, job_id):
         with qdb.sql_connection.TRN:
             job = _get_job(job_id)
 
+            if job.status != 'running':
+                raise HTTPError(
+                    403, "Can't complete job: not in a running state")
+
             payload = loads(self.request.body)
-            payload_success = payload['success']
-            if payload_success:
-                artifacts = payload['artifacts']
-                error = None
-            else:
-                artifacts = None
-                error = payload['error']
-            try:
-                job.complete(payload_success, artifacts, error)
-            except qdb.exceptions.QiitaDBOperationNotPermittedError as e:
-                raise HTTPError(403, str(e))
+            safe_submit(job.user.email, _job_completer, job_id, payload)
 
         self.finish()
 

qiita_db/handlers/tests/test_processing_job.py

@@ -17,6 +17,7 @@
 import numpy.testing as npt
 import pandas as pd
 
+from qiita_core.testing import wait_for_processing_job
 from qiita_db.handlers.tests.oauthbase import OauthTestingBase
 import qiita_db as qdb
 from qiita_db.handlers.processing_job import _get_job
@@ -169,6 +170,7 @@ def test_post_job_failure(self):
             '/qiita_db/jobs/bcc7ebcd-39c1-43e4-af2d-822e3589f14d/complete/',
             payload, headers=self.header)
         self.assertEqual(obs.code, 200)
+        wait_for_processing_job('bcc7ebcd-39c1-43e4-af2d-822e3589f14d')
         job = qdb.processing_job.ProcessingJob(
             'bcc7ebcd-39c1-43e4-af2d-822e3589f14d')
         self.assertEqual(job.status, 'error')
@@ -206,6 +208,7 @@ def test_post_job_success(self):
         obs = self.post(
             '/qiita_db/jobs/%s/complete/' % job.id,
             payload, headers=self.header)
+        wait_for_processing_job(job.id)
         self.assertEqual(obs.code, 200)
         self.assertEqual(job.status, 'success')
         self.assertEqual(qdb.util.get_count('qiita.artifact'),