Workflow object

qiita_db/artifact.py

@@ -376,7 +376,8 @@ def delete(cls, artifact_id):
                 raise qdb.exceptions.QiitaDBArtifactDeletionError(
                     artifact_id, "it has been submitted to VAMPS")
 
-            # Check if there is a job queued that will use the artifact
+            # Check if there is a job queued or running that will use/is using
+            # the artifact
             sql = """SELECT EXISTS(
                         SELECT *
                         FROM qiita.artifact_processing_job
@@ -412,6 +413,11 @@ def delete(cls, artifact_id):
                          WHERE processing_job_id IN %s"""
                 qdb.sql_connection.TRN.add(sql, [job_ids])
 
+            # Delete the entry from the artifact_output_processing_job table
+            sql = """DELETE FROM qiita.artifact_output_processing_job
+                     WHERE artifact_id = %s"""
+            qdb.sql_connection.TRN.add(sql, [artifact_id])
+
             # Detach the artifact from its filepaths
             sql = """DELETE FROM qiita.artifact_filepath
                      WHERE artifact_id = %s"""

qiita_db/handlers/processing_job.py

@@ -6,7 +6,6 @@
 # The full license is in the file LICENSE, distributed with this software.
 # -----------------------------------------------------------------------------
 
-from datetime import datetime
 from json import loads
 
 import qiita_db as qdb
@@ -104,15 +103,11 @@ def post(self, job_id):
         with qdb.sql_connection.TRN:
             job, success, error_msg = _get_job(job_id)
             if success:
-                job_status = job.status
-                if job_status == 'queued':
-                    job.status = 'running'
-                    job.heartbeat = datetime.now()
-                elif job_status == 'running':
-                    job.heartbeat = datetime.now()
-                else:
+                try:
+                    job.update_heartbeat_state()
+                except qdb.exceptions.QiitaDBOperationNotPermittedError as e:
                     success = False
-                    error_msg = 'Job already finished. Status: %s' % job_status
+                    error_msg = str(e)
 
         response = {'success': success, 'error': error_msg}
         self.write(response)
@@ -175,26 +170,19 @@ def post(self, job_id):
         with qdb.sql_connection.TRN:
             job, success, error_msg = _get_job(job_id)
             if success:
-                if job.status != 'running':
-                    success = False
-                    error_msg = 'Job in a non-running state.'
+                payload = loads(self.request.body)
+                payload_success = payload['success']
+                if payload_success:
+                    artifacts = payload['artifacts']
+                    error = None
                 else:
-                    payload = loads(self.request.body)
-                    if payload['success']:
-                        for artifact_data in payload['artifacts']:
-                            filepaths = artifact_data['filepaths']
-                            atype = artifact_data['artifact_type']
-                            parents = job.input_artifacts
-                            params = job.parameters
-                            qdb.artifact.Artifact.create(
-                                filepaths, atype, parents=parents,
-                                processing_parameters=params)
-                        job.status = 'success'
-                    else:
-                        log = qdb.logger.LogEntry.create(
-                            'Runtime', payload['error'])
-                        job.status = 'error'
-                        job.log = log
+                    artifacts = None
+                    error = payload['error']
+                try:
+                    job.complete(payload_success, artifacts, error)
+                except qdb.exceptions.QiitaDBOperationNotPermittedError as e:
+                    success = False
+                    error_msg = str(e)
 
         response = {'success': success, 'error': error_msg}
         self.write(response)

qiita_db/handlers/tests/test_processing_job.py

@@ -78,7 +78,7 @@ def test_post_job_already_finished(self):
             '', headers=self.header)
         self.assertEqual(obs.code, 200)
         exp = {'success': False,
-               'error': 'Job already finished. Status: success'}
+               'error': "Can't execute heartbeat on job: already completed"}
         self.assertEqual(loads(obs.body), exp)
 
     def test_post(self):
@@ -177,12 +177,13 @@ def test_post_job_does_not_exists(self):
         self.assertEqual(loads(obs.body), exp)
 
     def test_post_job_not_running(self):
-        payload = dumps({'sucess': False, 'error': 'Job failure'})
+        payload = dumps({'success': False, 'error': 'Job failure'})
         obs = self.post(
             '/qiita_db/jobs/063e553b-327c-4818-ab4a-adfe58e49860/complete/',
             payload, headers=self.header)
         self.assertEqual(obs.code, 200)
-        exp = {'success': False, 'error': "Job in a non-running state."}
+        exp = {'success': False,
+               'error': "Can't complete job: not in a running state"}
         self.assertEqual(loads(obs.body), exp)
 
     def test_post_job_failure(self):
@@ -209,10 +210,8 @@ def test_post_job_success(self):
         exp_artifact_count = qdb.util.get_count('qiita.artifact') + 1
         payload = dumps(
             {'success': True, 'error': '',
-             'artifacts': [
-                 {'filepaths': [(fp, 'biom')],
-                  'artifact_type': 'BIOM'}
-             ]})
+             'artifacts': {'OTU table': {'filepaths': [(fp, 'biom')],
+                                         'artifact_type': 'BIOM'}}})
         obs = self.post(
             '/qiita_db/jobs/bcc7ebcd-39c1-43e4-af2d-822e3589f14d/complete/',
             payload, headers=self.header)