Moving job completion to internal plugin structure

qiita_db/commands.py

@@ -9,8 +9,6 @@
 from functools import partial
 from future import standard_library
 from json import loads
-from sys import exc_info
-import traceback
 
 import qiita_db as qdb
 
@@ -270,27 +268,3 @@ def update_artifact_from_cmd(filepaths, filepath_types, artifact_id):
         qdb.sql_connection.TRN.execute()
 
     return artifact
-
-
-def complete_job_cmd(job_id, payload):
-    """Completes the given job
-
-    Parameters
-    ----------
-    job_id : str
-        The job id
-    payload : str
-        JSON string with the payload to complete the job
-    """
-    payload = loads(payload)
-    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()))

qiita_db/handlers/processing_job.py

@@ -7,7 +7,6 @@
 # -----------------------------------------------------------------------------
 
 from json import loads
-from multiprocessing import Process
 
 from tornado.web import HTTPError
 
@@ -46,27 +45,6 @@ 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 : str
-        The JSON string with the parameters of the HTTP POST request that is
-        completing the job
-    """
-    import qiita_db as qdb
-    qdb.sql_connection.TRN.close()
-
-    success, error = qdb.processing_job.private_job_submitter(
-        "Complete job %s" % job_id, 'complete_job', [job_id, payload])
-    if not success:
-        job = qdb.processing_job.ProcessingJob(job_id)
-        job.complete(False, error=error)
-
-
 class JobHandler(OauthBaseHandler):
     @authenticate_oauth
     def get(self, job_id):
@@ -159,10 +137,14 @@ def post(self, job_id):
                 raise HTTPError(
                     403, "Can't complete job: not in a running state")
 
-            p = Process(target=_job_completer,
-                        args=(job_id, self.request.body))
-            p.start()
-            # safe_submit(job.user.email, _job_completer, job_id, payload)
+            qiita_plugin = qdb.software.Software.from_name_and_version(
+                'Qiita', 'alpha')
+            cmd = qiita_plugin.get_command('complete_job')
+            params = qdb.software.Parameters.load(
+                cmd, values_dict={'job_id': job_id,
+                                  'payload': self.request.body})
+            job = qdb.processing_job.ProcessingJob.create(job.user, params)
+            job.submit()
 
         self.finish()
 

qiita_db/processing_job.py

@@ -63,7 +63,6 @@ def _job_submitter(job_id, cmd):
     cmd : str
         The command to execute the job
     """
-    qdb.sql_connection.TRN.close()
     job = ProcessingJob(job_id)
     std_out, std_err, return_value = _system_call(cmd)
     if return_value != 0:

qiita_db/support_files/patches/python_patches/58.py

@@ -67,3 +67,8 @@
     Command.create(qiita_plugin, "delete_sample_or_column",
                    "Deletes a sample or a columns from the metadata",
                    parameters)
+
+    # Create the command to complete a job
+    parameters = {'job_id': ['string', None], 'payload': ['string', None]}
+    Command.create(qiita_plugin, "complete_job", "Completes a given job",
+                   parameters)

qiita_db/test/test_commands.py

@@ -15,10 +15,8 @@
 from future import standard_library
 from functools import partial
 from operator import itemgetter
-from json import dumps
 
 import pandas as pd
-import numpy.testing as npt
 
 from qiita_core.util import qiita_test_checker
 
@@ -404,84 +402,6 @@ def test_update_artifact_from_cmd(self):
             self.assertEqual(qdb.util.compute_checksum(obs[1]), exp)
 
 
-@qiita_test_checker()
-class TestCompleteJobCmd(TestCase):
-    def setUp(self):
-        self._clean_up_files = []
-
-    def tearDown(self):
-        for fp in self._clean_up_files:
-            if exists(fp):
-                remove(fp)
-
-    def test_complete_success(self):
-        pt = npt.assert_warns(
-            qdb.exceptions.QiitaDBWarning,
-            qdb.metadata_template.prep_template.PrepTemplate.create,
-            pd.DataFrame({'new_col': {'1.SKD6.640190': 1}}),
-            qdb.study.Study(1), '16S')
-        job = qdb.processing_job.ProcessingJob.create(
-            qdb.user.User('test@foo.bar'),
-            qdb.software.Parameters.load(
-                qdb.software.Command.get_validator('BIOM'),
-                values_dict={'template': pt.id, 'files':
-                             dumps({'BIOM': ['file']}),
-                             'artifact_type': 'BIOM'}))
-        job._set_status('running')
-
-        fd, fp = mkstemp(suffix='_table.biom')
-        close(fd)
-        with open(fp, 'w') as f:
-            f.write('\n')
-
-        self._clean_up_files.append(fp)
-
-        exp_artifact_count = qdb.util.get_count('qiita.artifact') + 1
-        payload = dumps(
-            {'success': True, 'error': '',
-             'artifacts': {'OTU table': {'filepaths': [(fp, 'biom')],
-                                         'artifact_type': 'BIOM'}}})
-        qdb.commands.complete_job_cmd(job.id, payload)
-        self.assertEqual(job.status, 'success')
-        self.assertEqual(qdb.util.get_count('qiita.artifact'),
-                         exp_artifact_count)
-
-    def test_complete_job_error(self):
-        payload = dumps({'success': False, 'error': 'Job failure'})
-        qdb.commands.complete_job_cmd(
-            'bcc7ebcd-39c1-43e4-af2d-822e3589f14d', payload)
-        job = qdb.processing_job.ProcessingJob(
-            'bcc7ebcd-39c1-43e4-af2d-822e3589f14d')
-        self.assertEqual(job.status, 'error')
-        self.assertEqual(job.log,
-                         qdb.logger.LogEntry.newest_records(numrecords=1)[0])
-        self.assertEqual(job.log.msg, 'Job failure')
-
-    def test_complete_error(self):
-        pt = npt.assert_warns(
-            qdb.exceptions.QiitaDBWarning,
-            qdb.metadata_template.prep_template.PrepTemplate.create,
-            pd.DataFrame({'new_col': {'1.SKD6.640190': 1}}),
-            qdb.study.Study(1), '16S')
-        job = qdb.processing_job.ProcessingJob.create(
-            qdb.user.User('test@foo.bar'),
-            qdb.software.Parameters.load(
-                qdb.software.Command.get_validator('BIOM'),
-                values_dict={'template': pt.id, 'files':
-                             dumps({'BIOM': ['file']}),
-                             'artifact_type': 'BIOM'}))
-        job._set_status('running')
-
-        fp = '/surprised/if/this/path/exists.biom'
-        payload = dumps(
-            {'success': True, 'error': '',
-             'artifacts': {'OTU table': {'filepaths': [(fp, 'biom')],
-                                         'artifact_type': 'BIOM'}}})
-        qdb.commands.complete_job_cmd(job.id, payload)
-        self.assertEqual(job.status, 'error')
-        self.assertIn('No such file or directory', job.log.msg)
-
-
 CONFIG_1 = """[required]
 timeseries_type_id = 1
 metadata_complete = True

qiita_ware/private_plugin.py

@@ -6,7 +6,7 @@
 # The full license is in the file LICENSE, distributed with this software.
 # -----------------------------------------------------------------------------
 
-from json import dumps
+from json import dumps, loads
 from sys import exc_info
 from time import sleep
 from os import remove
@@ -273,6 +273,32 @@ def delete_sample_or_column(job):
         job._set_status('success')
 
 
+def complete_job(job):
+    """Deletes a sample or a column from the metadata
+
+    Parameters
+    ----------
+    job : qiita_db.processing_job.ProcessingJob
+        The processing job performing the task
+    """
+    with qdb.sql_connection.TRN:
+        param_vals = job.parameters.values
+        payload = loads(param_vals['payload'])
+        if payload['success']:
+            artifacts = payload['artifacts']
+            error = None
+        else:
+            artifacts = None
+            error = payload['error']
+        c_job = qdb.processing_job.ProcessingJob(param_vals['job_id'])
+        try:
+            c_job.complete(payload['success'], artifacts, error)
+        except:
+            c_job._set_error(traceback.format_exception(*exc_info()))
+
+        job._set_status('success')
+
+
 TASK_DICT = {'build_analysis_files': build_analysis_files,
              'release_validators': release_validators,
              'submit_to_VAMPS': submit_to_VAMPS,
@@ -283,7 +309,8 @@ def delete_sample_or_column(job):
              'update_sample_template': update_sample_template,
              'delete_sample_template': delete_sample_template,
              'update_prep_template': update_prep_template,
-             'delete_sample_or_column': delete_sample_or_column}
+             'delete_sample_or_column': delete_sample_or_column,
+             'complete_job': complete_job}
 
 
 def private_task(job_id):

qiita_ware/test/test_private_plugin.py

@@ -10,14 +10,14 @@
 from os.path import join, dirname, abspath, exists
 from os import close, remove
 from tempfile import mkstemp
-from json import loads
+from json import loads, dumps
 
 import pandas as pd
 import numpy.testing as npt
 
 from qiita_core.util import qiita_test_checker
 from qiita_core.qiita_settings import r_client
-from qiita_db.software import Software, Parameters
+from qiita_db.software import Software, Parameters, Command
 from qiita_db.processing_job import ProcessingJob
 from qiita_db.user import User
 from qiita_db.study import Study, StudyPerson
@@ -26,6 +26,8 @@
 from qiita_db.exceptions import QiitaDBWarning
 from qiita_db.artifact import Artifact
 from qiita_db.exceptions import QiitaDBUnknownIDError
+from qiita_db.util import get_count
+from qiita_db.logger import LogEntry
 from qiita_ware.private_plugin import private_task
 
 
@@ -301,6 +303,74 @@ def test_delete_sample_or_column(self):
         self.assertIn('Unknown value "unknown". Choose between "samples" '
                       'and "columns"', job.log.msg)
 
+    def test_complete_job(self):
+        # Complete success
+        pt = npt.assert_warns(
+            QiitaDBWarning, PrepTemplate.create,
+            pd.DataFrame({'new_col': {'1.SKD6.640190': 1}}),
+            Study(1), '16S')
+        c_job = ProcessingJob.create(
+            User('test@foo.bar'),
+            Parameters.load(
+                Command.get_validator('BIOM'),
+                values_dict={'template': pt.id,
+                             'files': dumps({'BIOM': ['file']}),
+                             'artifact_type': 'BIOM'}))
+        c_job._set_status('running')
+        fd, fp = mkstemp(suffix='_table.biom')
+        close(fd)
+        with open(fp, 'w') as f:
+            f.write('\n')
+        self._clean_up_files.append(fp)
+        exp_artifact_count = get_count('qiita.artifact') + 1
+        payload = dumps(
+            {'success': True, 'error': '',
+             'artifacts': {'OTU table': {'filepaths': [(fp, 'biom')],
+                                         'artifact_type': 'BIOM'}}})
+        job = self._create_job('complete_job', {'job_id': c_job.id,
+                                                'payload': payload})
+        private_task(job.id)
+        self.assertEqual(job.status, 'success')
+        self.assertEqual(c_job.status, 'success')
+        self.assertEqual(get_count('qiita.artifact'), exp_artifact_count)
+
+        # Complete job error
+        payload = dumps({'success': False, 'error': 'Job failure'})
+        job = self._create_job(
+            'complete_job', {'job_id': 'bcc7ebcd-39c1-43e4-af2d-822e3589f14d',
+                             'payload': payload})
+        private_task(job.id)
+        self.assertEqual(job.status, 'success')
+        c_job = ProcessingJob('bcc7ebcd-39c1-43e4-af2d-822e3589f14d')
+        self.assertEqual(c_job.status, 'error')
+        self.assertEqual(c_job.log, LogEntry.newest_records(numrecords=1)[0])
+        self.assertEqual(c_job.log.msg, 'Job failure')
+
+        # Complete internal error
+        pt = npt.assert_warns(
+            QiitaDBWarning, PrepTemplate.create,
+            pd.DataFrame({'new_col': {'1.SKD6.640190': 1}}),
+            Study(1), '16S')
+        c_job = ProcessingJob.create(
+            User('test@foo.bar'),
+            Parameters.load(
+                Command.get_validator('BIOM'),
+                values_dict={'template': pt.id,
+                             'files': dumps({'BIOM': ['file']}),
+                             'artifact_type': 'BIOM'}))
+        c_job._set_status('running')
+        fp = '/surprised/if/this/path/exists.biom'
+        payload = dumps(
+            {'success': True, 'error': '',
+             'artifacts': {'OTU table': {'filepaths': [(fp, 'biom')],
+                                         'artifact_type': 'BIOM'}}})
+        job = self._create_job('complete_job', {'job_id': c_job.id,
+                                                'payload': payload})
+        private_task(job.id)
+        self.assertEqual(job.status, 'success')
+        self.assertEqual(c_job.status, 'error')
+        self.assertIn('No such file or directory', c_job.log.msg)
+
 
 if __name__ == '__main__':
     main()