Merge pull request #3140 from antgonza/admin-jobs

WIP: adding admin direct job creation

Author: charles-cowart

qiita_db/exceptions.py

@@ -49,14 +49,14 @@ class QiitaDBArtifactCreationError(QiitaDBError):
     """Exception when creating an artifact"""
     def __init__(self, reason):
         super(QiitaDBArtifactCreationError, self).__init__()
-        self.args = ("Cannot create artifact: %s" % reason,)
+        self.args = (f"Cannot create artifact: {reason}",)
 
 
 class QiitaDBArtifactDeletionError(QiitaDBError):
     """Exception when deleting an artifact"""
     def __init__(self, a_id, reason):
         super(QiitaDBArtifactDeletionError, self).__init__()
-        self.args = ("Cannot delete artifact %d: %s" % (a_id, reason),)
+        self.args = (f"Cannot delete artifact {a_id}: {reason}",)
 
 
 class QiitaDBDuplicateError(QiitaDBError):

qiita_db/processing_job.py

@@ -1015,11 +1015,16 @@ def release(self):
                 # This job is resulting from a private job
                 parents = None
                 params = None
-                cmd_out_id = None
                 name = None
                 data_type = a_info['data_type']
-                analysis = qdb.analysis.Analysis(
-                    job.parameters.values['analysis'])
+                pvals = job.parameters.values
+                if 'analysis' in pvals:
+                    cmd_out_id = None
+                    analysis = qdb.analysis.Analysis(
+                        job.parameters.values['analysis'])
+                else:
+                    cmd_out_id = provenance['cmd_out_id']
+                    analysis = None
                 a_info = a_info['artifact_data']
             else:
                 # This job is resulting from a plugin job
@@ -1264,7 +1269,7 @@ def _complete_artifact_transformation(self, artifacts_data):
                         "is allowed, found %d" % len(templates))
                 elif len(templates) == 1:
                     template = templates.pop()
-                else:
+                elif self.input_artifacts:
                     # In this case we have 0 templates. What this means is that
                     # this artifact is being generated in the analysis pipeline
                     # All the artifacts included in the analysis pipeline
@@ -1297,6 +1302,9 @@ def _complete_artifact_transformation(self, artifacts_data):
                               'cmd_out_id': cmd_out_id,
                               'name': art_name}
 
+                if self.command.software.type == 'private':
+                    provenance['data_type'] = 'Job Output Folder'
+
                 # Get the validator command for the current artifact type and
                 # create a new job
                 # see also release_validators()

qiita_pet/handlers/admin_processing_job.py

@@ -0,0 +1,86 @@
+# -----------------------------------------------------------------------------
+# Copyright (c) 2014--, The Qiita Development Team.
+#
+# Distributed under the terms of the BSD 3-clause License.
+#
+# The full license is in the file LICENSE, distributed with this software.
+# -----------------------------------------------------------------------------
+
+from tornado.gen import coroutine
+from tornado.web import HTTPError
+
+from .base_handlers import BaseHandler
+from qiita_core.util import execute_as_transaction
+
+from qiita_db.software import Software
+
+from json import dumps
+
+
+class AdminProcessingJobBaseClass(BaseHandler):
+    def _check_access(self):
+        if self.current_user.level not in {'admin', 'dev'}:
+            raise HTTPError(403, reason="User %s doesn't have sufficient "
+                            "privileges to view error page" %
+                            self.current_user.email)
+
+        return self
+
+    def _get_private_software(self):
+        # skipping the internal Qiita plugin and only selecting private
+        # commands
+        private_software = [s for s in Software.iter()
+                            if s.name != 'Qiita' and s.type == 'private']
+
+        return private_software
+
+
+class AdminProcessingJob(AdminProcessingJobBaseClass):
+    @coroutine
+    @execute_as_transaction
+    def get(self):
+        self._check_access()
+
+        self.render("admin_processing_job.html",
+                    private_software=self._get_private_software())
+
+
+class AJAXAdminProcessingJobListing(AdminProcessingJobBaseClass):
+    @coroutine
+    @execute_as_transaction
+    def get(self):
+        self._check_access()
+        echo = self.get_argument('sEcho')
+        command_id = int(self.get_argument('commandId'))
+
+        jobs = []
+        for ps in self._get_private_software():
+            for cmd in ps.commands:
+                if cmd.id != command_id:
+                    continue
+
+                for job in cmd.processing_jobs:
+                    msg = '' if job.status != 'error' else job.log.msg
+                    msg = msg.replace('\n', '</br>')
+                    outputs = []
+                    if job.status == 'success':
+                        outputs = [[k, v.id] for k, v in job.outputs.items()]
+                    validator_jobs = [v.id for v in job.validator_jobs]
+
+                    if job.heartbeat is not None:
+                        heartbeat = job.heartbeat.strftime('%Y-%m-%d %H:%M:%S')
+                    else:
+                        heartbeat = 'N/A'
+
+                    jobs.append([job.id, job.command.name, job.status, msg,
+                                 outputs, validator_jobs, heartbeat,
+                                 job.parameters.values])
+        results = {
+            "sEcho": echo,
+            "recordsTotal": len(jobs),
+            "recordsFiltered": len(jobs),
+            "data": jobs
+        }
+
+        # return the json in compact form to save transmit size
+        self.write(dumps(results, separators=(',', ':')))

qiita_pet/handlers/api_proxy/processing.py

@@ -135,19 +135,27 @@ def workflow_handler_post_req(user_id, command_id, params):
          'message': str,
          'workflow_id': int}
     """
-    parameters = Parameters.load(Command(command_id), json_str=params)
-
     status = 'success'
     message = ''
     try:
-        wf = ProcessingWorkflow.from_scratch(User(user_id), parameters)
+        parameters = Parameters.load(Command(command_id), json_str=params)
     except Exception as exc:
         wf = None
         wf_id = None
         job_info = None
         status = 'error'
         message = str(exc)
 
+    if status == 'success':
+        try:
+            wf = ProcessingWorkflow.from_scratch(User(user_id), parameters)
+        except Exception as exc:
+            wf = None
+            wf_id = None
+            job_info = None
+            status = 'error'
+            message = str(exc)
+
     if wf is not None:
         # this is safe as we are creating the workflow for the first time
         # and there is only one node. Remember networkx doesn't assure order

qiita_pet/handlers/artifact_handlers/base_handlers.py

@@ -126,14 +126,17 @@ def artifact_summary_get_request(user, artifact_id):
     # Check if the artifact is editable by the given user
     study = artifact.study
     analysis = artifact.analysis
-    editable = study.can_edit(user) if study else analysis.can_edit(user)
+    if artifact_type == 'job-output-folder':
+        editable = False
+    else:
+        editable = study.can_edit(user) if study else analysis.can_edit(user)
 
     buttons = []
     btn_base = (
         '<button onclick="if (confirm(\'Are you sure you want to %s '
         'artifact id: {0}?\')) {{ set_artifact_visibility(\'%s\', {0}) }}" '
         'class="btn btn-primary btn-sm">%s</button>').format(artifact_id)
-    if not analysis:
+    if not analysis and artifact_type != 'job-output-folder':
         # If the artifact is part of a study, the buttons shown depend in
         # multiple factors (see each if statement for an explanation of those)
         if qiita_config.require_approval:

qiita_pet/handlers/study_handlers/processing.py

@@ -6,6 +6,7 @@
 # The full license is in the file LICENSE, distributed with this software.
 # -----------------------------------------------------------------------------
 from tornado.web import authenticated
+from json import loads, dumps
 
 from qiita_pet.handlers.base_handlers import BaseHandler
 from qiita_pet.handlers.api_proxy import (
@@ -29,9 +30,9 @@ class ListOptionsHandler(BaseHandler):
     @authenticated
     def get(self):
         command_id = self.get_argument("command_id")
-        artifact_id = self.get_argument("artifact_id")
+        artifact_id = self.get_argument("artifact_id", None)
         # if the artifact id has ':' it means that it's a job in construction
-        if ':' in artifact_id:
+        if artifact_id is not None and ':' in artifact_id:
             artifact_id = None
         self.write(list_options_handler_get_req(command_id, artifact_id))
 
@@ -48,6 +49,17 @@ class WorkflowHandler(BaseHandler):
     def post(self):
         command_id = self.get_argument('command_id')
         params = self.get_argument('params')
+
+        if self.request.files:
+            parameters = loads(params)
+            for k, v in self.request.files.items():
+                # [0] there is only one file -- this block is needed because
+                # 'body' is a byte and JSON doesn't know how to translate it
+                parameters[k] = {'body': v[0]['body'].decode("utf-8"),
+                                 'filename': v[0]['filename'],
+                                 'content_type': v[0]['content_type']}
+            params = dumps(parameters)
+
         self.write(workflow_handler_post_req(
             self.current_user.id, command_id, params))
 

qiita_pet/handlers/study_handlers/tests/test_processing.py

@@ -64,6 +64,12 @@ def test_get(self):
         self.assertEqual(response.code, 200)
         self.assertEqual(loads(response.body), exp)
 
+        # test that it works fine with no artifact_id
+        response = self.get('/study/process/commands/options/',
+                            {'command_id': '3'})
+        self.assertEqual(response.code, 200)
+        self.assertEqual(loads(response.body), exp)
+
 
 class TestJobAJAX(TestHandlerBase):
     def test_get(self):
@@ -137,5 +143,46 @@ def test_patch(self):
         self.assertEqual(loads(response.body), exp)
 
 
+class TestWorkflowHandler(TestHandlerBase):
+    def test_post(self):
+        # test error
+        response = self.post('/study/process/workflow/',
+                             {'command_id': '3', 'params': '{}'})
+        self.assertEqual(response.code, 200)
+        exp = {'status': 'error', 'workflow_id': None, 'job': None,
+               'message': "The provided JSON string doesn't encode a parameter"
+                          " set for command 3. Missing required parameter: "
+                          "input_data"}
+        self.assertDictEqual(loads(response.body), exp)
+
+        # test success
+        response = self.post('/study/process/workflow/',
+                             {'command_id': '3',
+                              'params': '{"input_data": 1}'})
+        self.assertEqual(response.code, 200)
+        obs = loads(response.body)
+        # we are going to copy the workflow_id/job information because we only
+        # care about the reply
+        exp = {'status': 'success', 'workflow_id': obs['workflow_id'],
+               'job': obs['job'], 'message': ''}
+        self.assertEqual(obs, exp)
+
+        # test with files
+        response = self.post('/study/process/workflow/',
+                             {'command_id': '3', 'params': '{"input_data": 3}',
+                              'files': '{"template": {"body": b""}}',
+                              'headers': {
+                                  'Content-Type': 'application/json',
+                                  'Origin': 'localhost'
+                              }, })
+        self.assertEqual(response.code, 200)
+        obs = loads(response.body)
+        # we are going to copy the workflow_id/job information because we only
+        # care about the reply
+        exp = {'status': 'success', 'workflow_id': obs['workflow_id'],
+               'job': obs['job'], 'message': ''}
+        self.assertEqual(obs, exp)
+
+
 if __name__ == "__main__":
     main()