Offload template processing from webserver

qiita_db/metadata_template/sample_template.py

@@ -137,7 +137,7 @@ def delete(cls, id_):
             has_prep_templates = qdb.sql_connection.TRN.execute_fetchlast()
             if has_prep_templates:
                 raise qdb.exceptions.QiitaDBError(
-                    "Sample template can not be erased because there are prep "
+                    "Sample template cannot be erased because there are prep "
                     "templates associated.")
 
             table_name = cls._table_name(id_)

qiita_pet/handlers/api_proxy/prep_template.py

@@ -7,20 +7,26 @@
 # -----------------------------------------------------------------------------
 from __future__ import division
 import warnings
-
 from os import remove
 from os.path import basename
+from json import loads
+
 from natsort import natsorted
+from moi import r_client
 
 from qiita_core.util import execute_as_transaction
 from qiita_pet.handlers.api_proxy.util import check_access, check_fp
+from qiita_ware.context import safe_submit
+from qiita_ware.dispatchable import update_prep_template
 from qiita_db.metadata_template.util import load_template_to_dataframe
 from qiita_db.util import convert_to_id, get_files_from_uploads_folders
 from qiita_db.study import Study
 from qiita_db.user import User
 from qiita_db.ontology import Ontology
 from qiita_db.metadata_template.prep_template import PrepTemplate
 
+PREP_TEMPLATE_KEY_FORMAT = 'prep_template_%s'
+
 
 def _get_ENA_ontology():
     """Returns the information of the ENA ontology
@@ -124,6 +130,23 @@ def prep_template_ajax_get_req(user_id, prep_id):
 
     ontology = _get_ENA_ontology()
 
+    job_id = r_client.get(PREP_TEMPLATE_KEY_FORMAT % prep_id)
+    if job_id:
+        redis_info = loads(r_client.get(job_id))
+        processing = redis_info['status_msg'] == 'Running'
+        if processing:
+            alert_type = 'info'
+            alert_msg = 'This prep template is currently being updated'
+        else:
+            alert_type = redis_info['return']['status']
+            alert_msg = redis_info['return']['message'].replace('\n', '</br>')
+    else:
+        processing = False
+        alert_type = ''
+        alert_msg = ''
+
+    editable = Study(study_id).can_edit(User(user_id)) and not processing
+
     return {'status': 'success',
             'message': '',
             'name': name,
@@ -136,8 +159,10 @@ def prep_template_ajax_get_req(user_id, prep_id):
             'ontology': ontology,
             'artifact_attached': artifact_attached,
             'study_id': study_id,
-            'editable': Study(study_id).can_edit(User(user_id)),
-            'data_type': pt.data_type()}
+            'editable': editable,
+            'data_type': pt.data_type(),
+            'alert_type': alert_type,
+            'alert_message': alert_msg}
 
 
 @execute_as_transaction
@@ -372,7 +397,7 @@ def prep_template_patch_req(user_id, req_op, req_path, req_value=None,
         if len(req_path) != 2:
             return {'status': 'error',
                     'message': 'Incorrect path parameter'}
-        prep_id = req_path[0]
+        prep_id = int(req_path[0])
         attribute = req_path[1]
 
         # Check if the user actually has access to the prep template
@@ -390,15 +415,8 @@ def prep_template_patch_req(user_id, req_op, req_path, req_value=None,
             if fp['status'] != 'success':
                 return fp
             fp = fp['file']
-            df = load_template_to_dataframe(fp)
-            with warnings.catch_warnings(record=True) as warns:
-                prep.extend(df)
-                prep.update(df)
-                remove(fp)
-
-                if warns:
-                    msg = '\n'.join(set(str(w.message) for w in warns))
-                    status = 'warning'
+            job_id = safe_submit(user_id, update_prep_template, prep_id, fp)
+            r_client.set(PREP_TEMPLATE_KEY_FORMAT % prep_id, job_id)
         else:
             # We don't understand the attribute so return an error
             return {'status': 'error',

qiita_pet/handlers/api_proxy/sample_template.py

@@ -6,25 +6,25 @@
 # The full license is in the file LICENSE, distributed with this software.
 # -----------------------------------------------------------------------------
 from __future__ import division
-import warnings
-
-from os import remove
+from json import loads
 
 from natsort import natsorted
+from moi import r_client
+
 from qiita_db.metadata_template.sample_template import SampleTemplate
 from qiita_db.exceptions import QiitaDBUnknownIDError
 from qiita_db.study import Study
 from qiita_core.util import execute_as_transaction
-from qiita_db.metadata_template.util import (load_template_to_dataframe,
-                                             looks_like_qiime_mapping_file)
+from qiita_db.metadata_template.util import looks_like_qiime_mapping_file
 from qiita_db.exceptions import QiitaDBColumnError
 from qiita_db.user import User
-
-from qiita_ware.metadata_pipeline import (
-    create_templates_from_qiime_mapping_file)
-from qiita_pet.util import convert_text_html
+from qiita_ware.dispatchable import (
+    create_sample_template, update_sample_template, delete_sample_template)
+from qiita_ware.context import safe_submit
 from qiita_pet.handlers.api_proxy.util import check_access, check_fp
 
+SAMPLE_TEMPLATE_KEY_FORMAT = 'sample_template_%s'
+
 
 def _check_sample_template_exists(samp_id):
     """Make sure a sample template exists in the system
@@ -214,38 +214,63 @@ def sample_template_summary_get_req(samp_id, user_id):
         Format {num_samples: value,
                 category: [(val1, count1), (val2, count2), ...], ...}
     """
-    exists = _check_sample_template_exists(int(samp_id))
-    if exists['status'] != 'success':
-        return exists
     access_error = check_access(samp_id, user_id)
     if access_error:
         return access_error
-    try:
-        template = SampleTemplate(int(samp_id))
-    except QiitaDBUnknownIDError as e:
-        return {'status': 'error',
-                'message': str(e)}
+
+    job_id = r_client.get(SAMPLE_TEMPLATE_KEY_FORMAT % samp_id)
+    if job_id:
+        redis_info = loads(r_client.get(job_id))
+        processing = redis_info['status_msg'] == 'Running'
+        if processing:
+            alert_type = 'info'
+            alert_msg = 'This sample template is currently being processed'
+        else:
+            alert_type = redis_info['return']['status']
+            alert_msg = redis_info['return']['message'].replace('\n', '</br>')
+    else:
+        processing = False
+        alert_type = ''
+        alert_msg = ''
+
+    exists = _check_sample_template_exists(int(samp_id))
+    if exists['status'] != 'success':
+        return {'status': 'success',
+                'message': '',
+                'num_samples': 0,
+                'num_columns': 0,
+                'editable': not processing,
+                'alert_type': alert_type,
+                'alert_message': alert_msg,
+                'stats': {}}
+
+    template = SampleTemplate(int(samp_id))
 
     df = template.to_dataframe()
+
+    editable = (Study(template.study_id).can_edit(User(user_id)) and not
+                processing)
+
     out = {'status': 'success',
            'message': '',
            'num_samples': df.shape[0],
            'num_columns': df.shape[1],
-           'editable': Study(template.study_id).can_edit(User(user_id)),
-           'summary': {}}
+           'editable': editable,
+           'alert_type': alert_type,
+           'alert_message': alert_msg,
+           'stats': {}}
 
     # drop the samp_id column if it exists
     if 'study_id' in df.columns:
         df.drop('study_id', axis=1, inplace=True)
     for column in df.columns:
         counts = df[column].value_counts()
-        out['summary'][str(column)] = [(str(key), counts[key])
-                                       for key in natsorted(counts.index)]
+        out['stats'][str(column)] = [(str(key), counts[key])
+                                     for key in natsorted(counts.index)]
 
     return out
 
 
-@execute_as_transaction
 def sample_template_post_req(study_id, user_id, data_type,
                              sample_template):
     """Creates the sample template from the given file
@@ -293,27 +318,13 @@ def sample_template_post_req(study_id, user_id, data_type,
                 'file': sample_template}
 
     study = Study(int(study_id))
-    try:
-        with warnings.catch_warnings(record=True) as warns:
-            if is_mapping_file:
-                create_templates_from_qiime_mapping_file(fp_rsp, study,
-                                                         data_type)
-            else:
-                SampleTemplate.create(load_template_to_dataframe(fp_rsp),
-                                      study)
-            remove(fp_rsp)
-
-            # join all the warning messages into one. Note that this
-            # info will be ignored if an exception is raised
-            if warns:
-                msg = '; '.join([convert_text_html(str(w.message))
-                                 for w in warns])
-                status = 'warning'
-    except Exception as e:
-        # Some error occurred while processing the sample template
-        # Show the error to the user so they can fix the template
-        status = 'error'
-        msg = str(e)
+
+    # Offload the creation of the sample template to the cluster
+    job_id = safe_submit(user_id, create_sample_template, fp_rsp, study,
+                         is_mapping_file, data_type)
+    # Store the job id attaching it to the sample template id
+    r_client.set(SAMPLE_TEMPLATE_KEY_FORMAT % study.id, job_id)
+
     return {'status': status,
             'message': msg,
             'file': sample_template}
@@ -358,23 +369,13 @@ def sample_template_put_req(study_id, user_id, sample_template):
 
     msg = ''
     status = 'success'
-    try:
-        with warnings.catch_warnings(record=True) as warns:
-            # deleting previous uploads and inserting new one
-            st = SampleTemplate(study_id)
-            df = load_template_to_dataframe(fp_rsp)
-            st.extend(df)
-            st.update(df)
-            remove(fp_rsp)
-
-            # join all the warning messages into one. Note that this info
-            # will be ignored if an exception is raised
-            if warns:
-                msg = '\n'.join(set(str(w.message) for w in warns))
-                status = 'warning'
-    except Exception as e:
-            status = 'error'
-            msg = str(e)
+
+    # Offload the update of the sample template to the cluster
+    job_id = safe_submit(user_id, update_sample_template, int(study_id),
+                         fp_rsp)
+    # Store the job id attaching it to the sample template id
+    r_client.set(SAMPLE_TEMPLATE_KEY_FORMAT % study_id, job_id)
+
     return {'status': status,
             'message': msg,
             'file': sample_template}
@@ -408,11 +409,12 @@ def sample_template_delete_req(study_id, user_id):
     if access_error:
         return access_error
 
-    try:
-        SampleTemplate.delete(int(study_id))
-    except Exception as e:
-        return {'status': 'error', 'message': str(e)}
-    return {'status': 'success'}
+    # Offload the deletion of the sample template to the cluster
+    job_id = safe_submit(user_id, delete_sample_template, int(study_id))
+    # Store the job id attaching it to the sample template id
+    r_client.set(SAMPLE_TEMPLATE_KEY_FORMAT % study_id, job_id)
+
+    return {'status': 'success', 'message': ''}
 
 
 @execute_as_transaction

qiita_pet/handlers/api_proxy/tests/test_prep_template.py

@@ -10,9 +10,12 @@
 from os.path import join, exists
 from string import ascii_letters
 from random import choice
+from time import sleep
+from json import loads
 
 import pandas as pd
 import numpy.testing as npt
+from moi import r_client
 
 from qiita_core.util import qiita_test_checker
 from qiita_db.artifact import Artifact
@@ -83,7 +86,9 @@ def test_prep_template_ajax_get_req(self):
                'artifact_attached': True,
                'study_id': 1,
                'editable': True,
-               'data_type': '18S'}
+               'data_type': '18S',
+               'alert_type': '',
+               'alert_message': ''}
         self.assertEqual(obs, exp)
 
         obs = prep_template_ajax_get_req('admin@foo.bar', 1)
@@ -276,6 +281,8 @@ def tearDown(self):
             with open(fp, 'w') as f:
                 f.write('')
 
+        r_client.flushdb()
+
     def test_prep_template_graph_get_req(self):
         obs = prep_template_graph_get_req(1, 'test@foo.bar')
         exp = {'edge_list': [(1, 3), (1, 2), (2, 4), (2, 5), (2, 6)],
@@ -392,15 +399,17 @@ def test_prep_template_patch_req(self):
         # Update prep template data
         obs = prep_template_patch_req(
             'test@foo.bar', 'replace', '/1/data', 'update.txt')
-        self.assertEqual(obs['status'], 'warning')
-        exp = [
-            'Sample names were already prefixed with the study id.',
-            'The following columns have been added to the existing template: '
-            'new_col',
-            'There are no differences between the data stored in the DB and '
-            'the new data provided']
-        self.assertItemsEqual(obs['message'].split('\n'), exp)
-        self.assertEqual(pt['1.SKD6.640190']['new_col'], 'new_value')
+        self.assertEqual(obs, exp)
+        obs = r_client.get('prep_template_1')
+        self.assertIsNotNone(obs)
+
+        # This is needed so the clean up works - this is a distributed system
+        # so we need to make sure that all processes are done before we reset
+        # the test database
+        redis_info = loads(r_client.get(obs))
+        while redis_info['status_msg'] == 'Running':
+            sleep(0.05)
+            redis_info = loads(r_client.get(obs))
 
     def test_prep_template_patch_req_errors(self):
         # Operation not supported