Transfer update delete templates (#2274)

* Moving update_sample_template

* Transfer update_sample_template

* Porting update prep template

* Moving delete sample or column

* Removing tests

* Removing dispatchable and its tests

* Updating interface to use the new functionality'

* Adapting the prep template GUI

* Submitting jobs

* Fixing tests

* Removing qiita_ware/context.py

* flake8ing

* Fixing _system_call

* Safeguarding the call to rollback

* Unmasking more errors

* Forcing different connections on different processes

* Moving job completion to internal plugin structure

* Removing unused code

* Forcing the creation of a new transaction on the jobs

* Fixing tests

* forcing the commit

* Fixing all tests

* Addressing @antgonza's comments

* Addressing @antgonza's comment

* Addressing @ElDeveloper's comments

Author: josenavas

.travis.yml

@@ -41,19 +41,17 @@ install:
   - mkdir ~/.qiita_plugins
   - cp $PWD/qiita_core/support_files/BIOM\ type_2.1.4.conf ~/.qiita_plugins
 before_script:
-    # Some of the tests rely on the plugin system to complete successfuly.
-    # Thus, we need a qiita webserver running to be able to execute the tests.
-  - export MOI_CONFIG_FP=`pwd`/qiita_core/support_files/config_test.cfg
   # EBI, see the end of before_install about why this block is commented out
   # - if [ ${TRAVIS_PULL_REQUEST} == "false" ]; then
   #   export QIITA_CONFIG_FP=`pwd`/qiita_core/support_files/config_test_travis.cfg;
-  #   export MOI_CONFIG_FP=`pwd`/qiita_core/support_files/config_test_travis.cfg;
   #   fi
   - qiita-env make --no-load-ontologies
   - qiita-test-install
 script:
-  - sleep 5
+  # Some of the tests rely on the plugin system to complete successfuly.
+  # Thus, we need a qiita webserver running to be able to execute the tests.
   - qiita pet webserver --no-build-docs start &
+  - sleep 5
   - QIITA_PID=$!
   - nosetests $COVER_PACKAGE --with-doctest --with-coverage --with-timer -v --cover-package=$COVER_PACKAGE
   - kill $QIITA_PID

qiita_core/testing.py

@@ -38,14 +38,7 @@ def wait_for_prep_information_job(prep_id, raise_if_none=True):
     if res is not None:
         payload = loads(res)
         job_id = payload['job_id']
-        if payload['is_qiita_job']:
-            wait_for_processing_job(job_id)
-        else:
-            redis_info = loads(r_client.get(job_id))
-            while redis_info['status_msg'] == 'Running':
-                sleep(0.5)
-                redis_info = loads(r_client.get(job_id))
-        sleep(0.5)
+        wait_for_processing_job(job_id)
 
 
 def wait_for_processing_job(job_id):

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/environment_manager.py

@@ -17,7 +17,7 @@
 from future.utils import viewitems
 
 from qiita_core.exceptions import QiitaEnvironmentError
-from qiita_core.qiita_settings import qiita_config
+from qiita_core.qiita_settings import qiita_config, r_client
 import qiita_db as qdb
 
 
@@ -318,6 +318,8 @@ def drop_and_rebuild_tst_database():
 
         qdb.sql_connection.TRN.execute()
 
+        r_client.flushdb()
+
 
 def reset_test_database(wrapped_fn):
     """Decorator that drops the qiita schema, rebuilds and repopulates the

qiita_db/handlers/processing_job.py

@@ -7,7 +7,6 @@
 # -----------------------------------------------------------------------------
 
 from json import loads
-from multiprocessing import Process
 
 from tornado.web import HTTPError
 
@@ -46,26 +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
-
-    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):
@@ -158,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

@@ -53,45 +53,23 @@ def _system_call(cmd):
     return stdout, stderr, return_value
 
 
-def _job_submitter(job, cmd):
+def _job_submitter(job_id, cmd):
     """Executes the commands `cmd` and updates the job in case of failure
 
     Parameters
     ----------
-    job : qiita_db.processing_job.ProcesingJob
-        The job that is executed by cmd
+    job_id : str
+        The job id that is executed by cmd
     cmd : str
         The command to execute the job
     """
     std_out, std_err, return_value = _system_call(cmd)
     if return_value != 0:
-        error = ("Error submitting job '%s':\nStd output:%s\nStd error:%s"
-                 % (job.id, std_out, std_err))
-        job.complete(False, error=error)
-
-
-def private_job_submitter(job_name, command, args):
-    """Submits a private job
-
-    Parameters
-    ----------
-    job_name : str
-        The name of the job
-    command: str
-        The private command to be executed
-    args: list of str
-        The arguments to the private command
-    """
-
-    cmd = "%s '%s' %s %s" % (qiita_config.private_launcher,
-                             qiita_config.qiita_env, command,
-                             ' '.join("'%s'" % a for a in args))
-    std_out, std_err, return_value = _system_call(cmd)
-    error = ""
-    if return_value != 0:
-        error = ("Can't submit private task '%s':\n"
-                 "Std output:%s\nStd error: %s" % (command, std_out, std_err))
-    return (return_value == 0), error
+        error = ("Error submitting job:\nStd output:%s\nStd error:%s"
+                 % (std_out, std_err))
+        # Forcing the creation of a new connection
+        qdb.sql_connection.create_new_transaction()
+        ProcessingJob(job_id).complete(False, error=error)
 
 
 class ProcessingJob(qdb.base.QiitaObject):
@@ -347,14 +325,19 @@ def submit(self):
         QiitaDBOperationNotPermittedError
             If the job is not in 'waiting' or 'in_construction' status
         """
-        status = self.status
-        if status not in {'in_construction', 'waiting'}:
-            raise qdb.exceptions.QiitaDBOperationNotPermittedError(
-                "Can't submit job, not in 'in_construction' or "
-                "'waiting' status. Current status: %s" % status)
-        self._set_status('queued')
+        with qdb.sql_connection.TRN:
+            status = self.status
+            if status not in {'in_construction', 'waiting'}:
+                raise qdb.exceptions.QiitaDBOperationNotPermittedError(
+                    "Can't submit job, not in 'in_construction' or "
+                    "'waiting' status. Current status: %s" % status)
+            self._set_status('queued')
+            # At this point we are going to involve other processes. We need
+            # to commit the changes to the DB or the other processes will not
+            # see these changes
+            qdb.sql_connection.TRN.commit()
         cmd = self._generate_cmd()
-        p = Process(target=_job_submitter, args=(self, cmd))
+        p = Process(target=_job_submitter, args=(self.id, cmd))
         p.start()
 
     def release(self):

qiita_db/sql_connection.py

@@ -646,9 +646,12 @@ def _raise_execution_error(self, sql, sql_args, error):
         """
         self.rollback()
 
-        raise ValueError(
-            "Error running SQL: %s. MSG: %s\n" % (
-                errorcodes.lookup(error.pgcode), error.message))
+        try:
+            ec_lu = errorcodes.lookup(error.pgcode)
+            raise ValueError(
+                "Error running SQL: %s. MSG: %s\n" % (ec_lu, error.message))
+        except KeyError:
+            raise ValueError("Error running SQL query: %s" % error.message)
 
     @_checker
     def add(self, sql, sql_args=None, many=False):
@@ -882,11 +885,13 @@ def rollback(self):
         # Reset the queries, the results and the index
         self._queries = []
         self._results = []
-        try:
-            self._connection.rollback()
-        except Exception:
-            self._connection.close()
-            raise
+
+        if self._connection is not None and self._connection.closed == 0:
+            try:
+                self._connection.rollback()
+            except Exception:
+                self._connection.close()
+                raise
         # Execute the post rollback functions
         self._funcs_executor(self._post_rollback_funcs, "rollback")
 
@@ -937,3 +942,12 @@ def add_post_rollback_func(self, func, *args, **kwargs):
 
 # Singleton pattern, create the transaction for the entire system
 TRN = Transaction()
+
+
+def create_new_transaction():
+    """Creates a new global transaction
+
+    This is needed when using multiprocessing
+    """
+    global TRN
+    TRN = Transaction()

qiita_db/support_files/patches/python_patches/58.py

@@ -14,24 +14,24 @@
     qiita_plugin = Software.from_name_and_version('Qiita', 'alpha')
 
     # Create the submit command for VAMPS command
-    parameters = {'artifact': ['artifact:["Demultiplexed"]', None]}
+    parameters = {'artifact': ['integer', None]}
     Command.create(qiita_plugin, "submit_to_VAMPS",
                    "submits an artifact to VAMPS", parameters)
 
     # Create the copy artifact command
-    parameters = {'artifact': ['artifact:["Demultiplexed"]', None],
+    parameters = {'artifact': ['integer', None],
                   'prep_template': ['prep_template', None]}
     Command.create(qiita_plugin, "copy_artifact",
                    "Creates a copy of an artifact", parameters)
 
     # Create the submit command for EBI command
-    parameters = {'artifact': ['artifact:["Demultiplexed"]', None],
+    parameters = {'artifact': ['integer', None],
                   'submission_type': ['choice:["ADD", "MODIFY"]', 'ADD']}
     Command.create(qiita_plugin, "submit_to_EBI",
                    "submits an artifact to EBI", parameters)
 
     # Create the submit command for delete_artifact
-    parameters = {'artifact': ['artifact:["Demultiplexed"]', None]}
+    parameters = {'artifact': ['integer', None]}
     Command.create(qiita_plugin, "delete_artifact",
                    "Delete an artifact", parameters)
 
@@ -41,3 +41,34 @@
         'is_mapping_file': ['boolean', True], 'data_type': ['string', None]}
     Command.create(qiita_plugin, "create_sample_template",
                    "Create a sample template", parameters)
+
+    # Create the update sample template command
+    parameters = {'study': ['integer', None], 'template_fp': ['string', None]}
+    Command.create(qiita_plugin, "update_sample_template",
+                   "Updates the sample template", parameters)
+
+    # Create the delete sample template command
+    parameters = {'study': ['integer', None]}
+    Command.create(qiita_plugin, "delete_sample_template",
+                   "Deletes a sample template", parameters)
+
+    # Create the update prep template command
+    parameters = {'prep_template': ['integer', None],
+                  'template_fp': ['string', None]}
+    Command.create(qiita_plugin, "update_prep_template",
+                   "Updates the prep template", parameters)
+
+    # Create the delete sample or column command
+    parameters = {
+        'obj_class': ['choice:["SampleTemplate", "PrepTemplate"]', None],
+        'obj_id': ['integer', None],
+        'sample_or_col': ['choice:["samples", "columns"]', None],
+        'name': ['string', None]}
+    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)