Offloading sample template processing from the server

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,64 @@ 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']
+    else:
+        processing = False
+        alert_type = ''
+        alert_msg = ''
+    alert_msg = alert_msg.replace('\n', '</br>')
+
+    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 +319,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 +370,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,10 +410,11 @@ 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)}
+    # 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'}
 
 

qiita_pet/handlers/study_handlers/sample_template.py

@@ -82,14 +82,18 @@ def get(self):
                        if download['status'] == 'success' else None)
 
         stats = sample_template_summary_get_req(study_id, self.current_user.id)
-        summary = stats['summary'] if 'summary' in stats else {}
-        num_samples = stats['num_samples'] if 'num_samples' in stats else 0
-        num_columns = stats['num_columns'] if 'num_columns' in stats else 0
-        editable = stats['editable'] if 'editable' in stats else True
-        self.render('study_ajax/sample_summary.html', stats=summary,
-                    num_samples=num_samples, num_columns=num_columns,
-                    download_id=download_id, files=files, study_id=study_id,
-                    data_types=data_types, editable=editable)
+        if stats['status'] != 'success':
+            if 'does not exist' in stats['message']:
+                raise HTTPError(404, stats['message'])
+            if 'User does not have access to study' in stats['message']:
+                raise HTTPError(403, stats['message'])
+
+        stats['download_id'] = download_id
+        stats['files'] = files
+        stats['study_id'] = study_id
+        stats['data_types'] = data_types
+
+        self.render('study_ajax/sample_summary.html', **stats)
 
     @authenticated
     def post(self):

qiita_pet/templates/study_ajax/sample_summary.html

@@ -47,6 +47,12 @@
         $('#file-btn').prop('disabled', false);
       }
     });
+
+    {% if alert_type != 'success' and alert_message != '' %}
+      bootstrapAlert("{% raw alert_message %}", "{{alert_type}}");
+    {% else %}
+      $('#bootstrap-alert').alert('close');
+    {% end %}
   });
 </script>
 

qiita_ware/dispatchable.py

@@ -42,3 +42,128 @@ def create_raw_data(filetype, prep_template, filepaths):
 def copy_raw_data(prep_template, artifact_id):
     """Creates a new raw data by copying from artifact_id"""
     Artifact.copy(Artifact(artifact_id), prep_template)
+
+
+def create_sample_template(fp, study, is_mapping_file, data_type):
+    """Creates a sample template
+
+    Parameters
+    ----------
+    fp : str
+        The file path to the template file
+    study : qiita_db.study.Study
+        The study to add the sample template to
+    is_mapping_file : bool
+        Whether `fp` contains a mapping file or a sample template
+    data_type : str
+        If `is_mapping_file` is True, the data type of the prep template to be
+        created
+
+    Returns
+    -------
+    dict of {str: str}
+        A dict of the form {'status': str, 'message': str}
+    """
+    # The imports need to be in here because this code is executed in
+    # the ipython workers
+    import warnings
+    from os import remove
+    from qiita_db.metadata_template.sample_template import SampleTemplate
+    from qiita_db.metadata_template.util import load_template_to_dataframe
+    from qiita_ware.metadata_pipeline import (
+        create_templates_from_qiime_mapping_file)
+
+    status = 'success'
+    msg = ''
+    try:
+        with warnings.catch_warnings(record=True) as warns:
+            if is_mapping_file:
+                create_templates_from_qiime_mapping_file(fp, study,
+                                                         data_type)
+            else:
+                SampleTemplate.create(load_template_to_dataframe(fp),
+                                      study)
+            remove(fp)
+
+            # 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:
+        # Some error occurred while processing the sample template
+        # Show the error to the user so they can fix the template
+        status = 'danger'
+        msg = str(e)
+
+    return {'status': status, 'message': msg}
+
+
+def update_sample_template(study_id, fp):
+    """Updates a sample template
+
+    Parameters
+    ----------
+    study_id : int
+        Study id whose template is going to be updated
+    fp : str
+        The file path to the template file
+
+    Returns
+    -------
+    dict of {str: str}
+        A dict of the form {'status': str, 'message': str}
+    """
+    import warnings
+    from os import remove
+    from qiita_db.metadata_template.util import load_template_to_dataframe
+    from qiita_db.metadata_template.sample_template import SampleTemplate
+
+    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)
+            st.extend(df)
+            st.update(df)
+            remove(fp)
+
+            # 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 = 'danger'
+            msg = str(e)
+
+    return {'status': status, 'message': msg}
+
+
+def delete_sample_template(study_id):
+    """Delete a sample template
+
+    Parameters
+    ----------
+    study_id : int
+        Study id whose template is going to be deleted
+
+    Returns
+    -------
+    dict of {str: str}
+        A dict of the form {'status': str, 'message': str}
+    """
+    from qiita_db.metadata_template.sample_template import SampleTemplate
+
+    msg = ''
+    status = 'success'
+    try:
+        SampleTemplate.delete(study_id)
+    except Exception as e:
+        status = 'danger'
+        msg = str(e)
+
+    return {'status': status, 'message': msg}