Workflow object

qiita_db/artifact.py

@@ -359,7 +359,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
@@ -395,6 +396,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.execute_heartbeat()
+                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.'
-                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'
+                payload = loads(self.request.body)
+                try:
+                    payload_success = payload['success']
+                    if payload_success:
+                        artifacts = payload['artifacts']
+                        error = None
                     else:
-                        log = qdb.logger.LogEntry.create(
-                            'Runtime', payload['error'])
-                        job.status = 'error'
-                        job.log = log
+                        artifacts = None
+                        error = payload['error']
+                    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)

qiita_db/processing_job.py

@@ -7,6 +7,9 @@
 # -----------------------------------------------------------------------------
 
 from uuid import UUID
+from datetime import datetime
+
+from future.utils import viewitems
 
 import qiita_db as qdb
 
@@ -186,8 +189,7 @@ def status(self):
             qdb.sql_connection.TRN.add(sql, [self.id])
             return qdb.sql_connection.TRN.execute_fetchlast()
 
-    @status.setter
-    def status(self, value):
+    def _set_status(self, value):
         """Sets the status of the job
 
         Parameters
@@ -219,6 +221,60 @@ def status(self, value):
             qdb.sql_connection.TRN.add(sql, [new_status, self.id])
             qdb.sql_connection.TRN.execute()
 
+    def complete(self, success, artifacts_data=None, error=None):
+        """Completes the job, either with a success or error status
+
+        Parameters
+        ----------
+        success : bool
+            Whether the job has completed successfully or not
+        artifacts_data : dict of dicts
+            The generated artifact information, keyed by output name.
+            The format of each of the internal dictionaries must be
+            {'filepaths': list of (str, str), 'artifact_type': str}
+            where `filepaths` contains the list of filepaths and filepath types
+            for the artifact and `artifact_type` the type of the artifact
+        error : str, optional
+            If the job was not successful, the error message
+
+        Raises
+        ------
+        qiita_db.exceptions.QiitaDBOperationNotPermittedError
+            If the job is not in running state
+        """
+        with qdb.sql_connection.TRN:
+            if self.status != 'running':
+                raise qdb.exceptions.QiitaDBOperationNotPermittedError(
+                    "Can't complete job: not in a running state")
+            if success:
+                if artifacts_data:
+                    artifact_ids = []
+                    for out_name, a_data in viewitems(artifacts_data):
+                        filepaths = a_data['filepaths']
+                        atype = a_data['artifact_type']
+                        parents = self.input_artifacts
+                        params = self.parameters
+                        a = qdb.artifact.Artifact.create(
+                            filepaths, atype, parents=parents,
+                            processing_parameters=params)
+                        cmd_out_id = qdb.util.convert_to_id(
+                            out_name, "command_output", "name")
+                        artifact_ids.append((cmd_out_id, a.id))
+                    if artifact_ids:
+                        sql = """INSERT INTO
+                                    qiita.artifact_output_processing_job
+                                    (artifact_id, processing_job_id,
+                                     command_output_id)
+                                 VALUES (%s, %s, %s)"""
+                        sql_params = [[aid, self.id, out_id]
+                                      for out_id, aid in artifact_ids]
+                        qdb.sql_connection.TRN.add(sql, sql_params, many=True)
+                self._set_status('success')
+            else:
+                log = qdb.logger.LogEntry.create('Runtime', error)
+                self._set_status('error')
+                self._set_log(log)
+
     @property
     def log(self):
         """The log entry attached to the job if it failed
@@ -240,8 +296,7 @@ def log(self):
                 res = qdb.logger.LogEntry(log_id)
         return res
 
-    @log.setter
-    def log(self, value):
+    def _set_log(self, value):
         """Attaches a log entry to the job
 
         Parameters
@@ -281,20 +336,28 @@ def heartbeat(self):
             qdb.sql_connection.TRN.add(sql, [self.id])
             return qdb.sql_connection.TRN.execute_fetchlast()
 
-    @heartbeat.setter
-    def heartbeat(self, value):
-        """Sets the timestamp for the last received heartbeat
+    def execute_heartbeat(self):
+        """Updates the heartbeat of the job
 
-        Parameters
-        ----------
-        value : datetime
-            The new timestamp
+        In case that the job is in `queued` status, it changes the status to
+        `running`.
+
+        Raises
+        ------
+        QiitaDBOperationNotPermittedError
+            If the job is already completed
         """
         with qdb.sql_connection.TRN:
+            status = self.status
+            if status == 'queued':
+                self._set_status('running')
+            elif status != 'running':
+                raise qdb.exceptions.QiitaDBOperationNotPermittedError(
+                    "Can't execute heartbeat on job: already completed")
             sql = """UPDATE qiita.processing_job
                      SET heartbeat = %s
                      WHERE processing_job_id = %s"""
-            qdb.sql_connection.TRN.add(sql, [value, self.id])
+            qdb.sql_connection.TRN.add(sql, [datetime.now(), self.id])
             qdb.sql_connection.TRN.execute()
 
     @property