feat(logic,views): RecombinantFieldError;

- `KeyError` -> `RecombinantFieldError`
- Sysadmin only refresh.

Author: JVickery-TBS

ckanext/recombinant/errors.py

@@ -7,6 +7,10 @@ class RecombinantException(Exception):
     pass
 
 
+class RecombinantFieldError(Exception):
+    pass
+
+
 class RecombinantConfigurationError(Exception):
     pass
 

ckanext/recombinant/templates/recombinant/force_update.html

@@ -0,0 +1,21 @@
+<h3>{{_('Errors in Recombinant Resource')}}</h3>
+<p>{{ _('The Recombinant resource is out of date:') }}</p>
+<ul>
+    <li>{{ _('Field(s) <em>{0}</em> do not exist in the database for this resource.').format(key_errors) }}</li>
+</ul>
+{% if not g.userobj.sysadmin %}
+  <p>{{ _('Please contact <a href="mailto:open-ouvert@tbs-sct.gc.ca">open-ouvert@tbs-sct.gc.ca</a> for assistance.') }}</p>
+{% else %}
+  {# only let sysadmins refresh the recombinant record via UI #}
+  <div class="clearfix"></div>
+  <h3>{{ _('Refresh Recombinant Resource') }}{% snippet 'snippets/sysadmin_only.html' %}</h3>
+  <p>{{ _('You can refresh your resource in the database to try to solve the problem.') }}</p>
+  <div class="clearfix"></div>
+  <p class="form-actions">
+    <form id="recombinant-refresh-form" action="{{ h.url_for('recombinant.refresh_dataset', resource_name=res_name, owner_org=owner_org) }}" method="post">
+      {{ h.csrf_input() }}
+      <input type="hidden" name="dataset_id" value="{{ dataset_id }}"/>
+      <button class="btn btn-danger mrgn-bttm-md m-b-3" type="submit" name="refresh" >{{ _('Refresh…') }}</button>
+    </form>
+  </p>
+{% endif %}

ckanext/recombinant/templates/recombinant/snippets/outdated_error.html

@@ -32,6 +32,7 @@
 
 from ckanext.recombinant.errors import (
     RecombinantException,
+    RecombinantFieldError,
     BadExcelData,
     format_trigger_error
 )
@@ -106,13 +107,15 @@ def upload(id: str) -> Union[Response, str]:
         return h.redirect_to('recombinant.preview_table',
                              resource_name=resource_name,
                              owner_org=org['name'])
-    except KeyError as e:
-        return render(
-            'recombinant/force_update.html',
-            extra_vars={'key_errors': str(e).replace("'", ''),
-                        'dataset_id': dataset['id'],
-                        'res_name': resource_name,
-                        'owner_org': dataset['owner_org']})
+    except RecombinantFieldError as e:
+        h.flash_error(render('recombinant/snippets/outdated_error.html',
+                             extra_vars={'key_errors': str(e).replace("'", ''),
+                                         'dataset_id': dataset['id'],
+                                         'res_name': resource_name,
+                                         'owner_org': dataset['owner_org']}))
+        return h.redirect_to('recombinant.preview_table',
+                             resource_name=resource_name,
+                             owner_org=org['name'])
     except BadExcelData as e:
         h.flash_error(_(e.message))
         return h.redirect_to('recombinant.preview_table',
@@ -331,13 +334,15 @@ def template(dataset_type: str, lang: str, owner_org: str) -> Union[Response, st
 
         try:
             append_data(book, record_data, chromo)
-        except KeyError as e:
-            return render(
-                'recombinant/force_update.html',
-                extra_vars={'key_errors': str(e).replace("'", ''),
-                            'dataset_id': dataset['id'],
-                            'res_name': resource['name'],
-                            'owner_org': dataset['owner_org']})
+        except RecombinantFieldError as e:
+            h.flash_error(render('recombinant/snippets/outdated_error.html',
+                                 extra_vars={'key_errors': str(e).replace("'", ''),
+                                             'dataset_id': dataset['id'],
+                                             'res_name': resource_name,
+                                             'owner_org': dataset['owner_org']}))
+            return h.redirect_to('recombinant.preview_table',
+                                 resource_name=resource_name,
+                                 owner_org=org['name'])
 
         resource_names = dict((r['id'], r['name']) for r in dataset['resources'])
         ds_info = lc.action.datastore_info(id=resource['id'])
@@ -665,13 +670,15 @@ def preview_table(resource_name: str,
 @recombinant.route('/recombinant/refresh/<resource_name>/<owner_org>',
                    methods=['GET', 'POST'])
 def refresh_dataset(resource_name: str, owner_org: str):
+    if not is_sysadmin(g.user):
+        return abort(403)
     if request.method != 'POST':
         # handle page refreshes
         h.flash_notice(_('Form not submitted, please try again.'))
         return h.redirect_to('recombinant.preview_table',
                              resource_name=resource_name,
                              owner_org=owner_org)
-    if 'confirm' in request.form:
+    if 'refresh' in request.form:
         dataset_id = request.form['dataset_id']
         if not dataset_id:
             h.flash_error(_('Could not determine dataset to update.'))
@@ -832,7 +839,7 @@ def _process_upload_file(lc: LocalCKAN,
                         key = key.group(1)
                     else:
                         key = _('unknown')
-                    raise KeyError(key)
+                    raise RecombinantFieldError(key)
                 else:
                     # type_ignore_reason: incomplete typing
                     pgerror = e.error_dict['records'][0]  # type: ignore

ckanext/recombinant/views.py

@@ -21,7 +21,7 @@
 from typing import Any, Dict, List, Tuple, Optional, Union
 
 from ckanext.recombinant.tables import get_geno
-from ckanext.recombinant.errors import RecombinantException
+from ckanext.recombinant.errors import RecombinantException, RecombinantFieldError
 from ckanext.recombinant.datatypes import datastore_type
 from ckanext.recombinant.helpers import (
     recombinant_choice_fields, recombinant_language_text)
@@ -175,6 +175,8 @@ def append_data(book: Workbook,
     current_row = DATA_FIRST_ROW
     for record in record_data:
         for col_num, field in template_cols_fields(chromo):
+            if field['datastore_id'] not in record:
+                raise RecombinantFieldError(field['datastore_id'])
             item = datastore_type_format(
                 record[field['datastore_id']], field['datastore_type'])
             sheet.cell(row=current_row, column=col_num).value = item