Merge pull request #140 from open-data/feature/sysadmin-ds-dump

DataStore Dump & DataStore Field Sync

Author: JVickery-TBS

changes/140.changes

@@ -0,0 +1 @@
+`recombinant_update` with `force` and `delete_fields` will now DROP old columns from the DataStore for fields that do not exist in the Schema.

changes/140.feature

@@ -0,0 +1 @@
+Added the DataStore Dump button for sysadmins to the Recombinant edit template.

ckanext/recombinant/cli.py

@@ -116,19 +116,26 @@ def remove_empty(dataset_type: Optional[List[str]] = None,
     is_flag=True,
     help="Force update of tables (required for changes to only primary keys/indexes)",
 )
+@click.option(
+    "-d",
+    "--delete-fields",
+    is_flag=True,
+    help="Deletes fields that are no longer in the Schema (requires --force-update)",
+)
 @click.option('-v', '--verbose', is_flag=True,
               type=click.BOOL, help='Increase verbosity.')
 def update(dataset_type: Optional[List[str]] = None,
            all_types: bool = False,
            force_update: bool = False,
+           delete_fields: bool = False,
            verbose: bool = False):
     """
     Triggers recombinant update for recombinant resources
 
     Full Usage:\n
         recombinant update (-a | DATASET_TYPE ...) [-f]
     """
-    _update(dataset_type, all_types, force_update, verbose=verbose)
+    _update(dataset_type, all_types, force_update, delete_fields, verbose=verbose)
 
 
 @recombinant.command(short_help="Delete recombinant datasets and all their data.")
@@ -349,6 +356,7 @@ def _show(dataset_type: Optional[str],
 def _update(dataset_types: Optional[List[str]],
             all_types: bool = False,
             force_update: bool = False,
+            delete_fields: bool = False,
             verbose: bool = False):
     """
     Triggers recombinant update for recombinant resources
@@ -366,7 +374,8 @@ def _update(dataset_types: Optional[List[str]],
                 click.echo('%s %s updating' % (dtype, o))
                 lc.action.recombinant_update(
                     owner_org=o, dataset_type=dtype,
-                    force_update=force_update)
+                    force_update=force_update,
+                    delete_fields=delete_fields)
 
 
 def _expand_dataset_types(dataset_types: Optional[List[str]],
@@ -471,7 +480,12 @@ def _load_one_csv_file(name: str) -> int:
     _path, csv_name = os.path.split(name)
     assert csv_name.endswith('.csv'), csv_name
     resource_name = csv_name[:-4]
-    click.echo(resource_name)
+    singular_org_name = None
+    if '.' in resource_name:
+        singular_org_name, resource_name = tuple(resource_name.split('.'))
+    click.echo('Resource name: %s' % resource_name)
+    if singular_org_name:
+        click.echo('Organization name: %s' % singular_org_name)
     chromo = get_chromo(resource_name)
 
     dataset_type = chromo['dataset_type']
@@ -480,6 +494,11 @@ def _load_one_csv_file(name: str) -> int:
     errors = 0
 
     for org_name, records in csv_data_batch(name, chromo):
+        if not org_name and not singular_org_name:
+            click.echo('could not find any org!')
+            return 1
+        if not org_name and singular_org_name:
+            org_name = singular_org_name
         results = lc.action.package_search(
             q='type:%s AND organization:%s' % (dataset_type, org_name),
             include_private=True,

ckanext/recombinant/logic.py

@@ -62,6 +62,8 @@ def recombinant_update(context: Context, data_dict: DataDict):
     :param owner_org: organization name or id
     :param delete_resources: True to delete extra resources found
     :param force_update: True to force updating of datastore tables
+    :param delete_fields: True to delete old fields not in schema,
+                          requires force_update=True
     '''
     lc, geno, dataset = _action_get_dataset(context, data_dict)
 
@@ -70,7 +72,8 @@ def recombinant_update(context: Context, data_dict: DataDict):
         delete_resources=asbool(data_dict.get('delete_resources', False)))
     _update_datastore(
         lc, geno, dataset,
-        force_update=asbool(data_dict.get('force_update', False)))
+        force_update=asbool(data_dict.get('force_update', False)),
+        delete_fields=asbool(data_dict.get('delete_fields', False)))
 
 
 def recombinant_show(context: Context, data_dict: DataDict) -> Dict[str, Any]:
@@ -113,11 +116,12 @@ def recombinant_show(context: Context, data_dict: DataDict) -> Dict[str, Any]:
         try:
             ds = lc.action.datastore_search(
                 resource_id=resource['id'],
-                limit=1)
+                limit=0)
             datastore_correct = _datastore_match(r['fields'], ds['fields'])
             out['datastore_correct'] = datastore_correct
             resources_correct = resources_correct and datastore_correct
             out['datastore_rows'] = ds.get('total', 0)
+            out['datastore_active'] = True
         except NotFound:
             out['error'] = 'datastore table missing'
             resources_correct = False
@@ -244,7 +248,8 @@ def _update_dataset(lc: LocalCKAN,
 def _update_datastore(lc: LocalCKAN,
                       geno: Dict[str, Any],
                       dataset: Dict[str, Any],
-                      force_update: bool = False):
+                      force_update: bool = False,
+                      delete_fields: bool = False):
     """
     call lc.action.datastore_create to create tables or add
     columns to existing datastore tables based on dataset definition
@@ -259,6 +264,7 @@ def _update_datastore(lc: LocalCKAN,
             chromo['resource_name'], dataset['id'])
         resource_id = resource_ids[chromo['resource_name']]
         fields = datastore_fields(chromo['fields'], datastore_text_types)
+        do_delete_fields = False
         try:
             ds = lc.action.datastore_search(resource_id=resource_id, limit=0)
         except NotFound:
@@ -273,6 +279,18 @@ def _update_datastore(lc: LocalCKAN,
             for f in datastore_fields(chromo['fields'], datastore_text_types):
                 if f['id'] not in seen:
                     fields.append(f)
+            if delete_fields:
+                # remove any fields from DS not in Schema
+                new_fields = []
+                schema_field_ids = set(
+                    f['id'] for f in datastore_fields(chromo['fields'],
+                                                      datastore_text_types))
+                for f in fields:
+                    if f['id'] not in schema_field_ids:
+                        do_delete_fields = True
+                        continue
+                    new_fields.append(f)
+                fields = new_fields
 
         trigger_names = _update_triggers(lc, chromo)
 
@@ -292,6 +310,7 @@ def _update_datastore(lc: LocalCKAN,
         lc.action.datastore_create(
             resource_id=resource_id,
             fields=fields,
+            delete_fields=do_delete_fields,
             primary_key=chromo.get('datastore_primary_key', []),
             foreign_keys=foreign_keys,
             indexes=chromo.get('datastore_indexes', []),
@@ -410,7 +429,8 @@ def _datastore_match(fs: List[Dict[str, Any]], fields: List[Dict[str, Any]]) ->
     """
     # XXX: does not check types or extra columns at this time
     existing = set(c['id'] for c in fields)
-    return all(f['datastore_id'] in existing for f in fs)
+    return all(f['datastore_id'] in existing for f in fs
+               if not f.get('published_resource_computed_field', False))
 
 
 @chained_action

ckanext/recombinant/read_csv.py

@@ -39,8 +39,7 @@ def csv_data_batch(csv_path: str,
 
         if strict:
             expected = [f['datastore_id'] for f in chromo['fields'] if not f.get(
-                'published_resource_computed_field')] + \
-                ['owner_org', 'owner_org_title']
+                'published_resource_computed_field')]
             assert cols == expected, 'column mismatch:\n{0}\n{1}'.format(
                 cols, expected)
 

ckanext/recombinant/templates/recombinant/resource_edit.html

@@ -45,6 +45,17 @@
   {% block action_panels %}
     {% if dataset %}
       {% if 'error' not in resource %}
+        {% if g.userobj.sysadmin and not resource.datastore_correct %}
+          {# only let sysadmins refresh the recombinant record via UI #}
+          <div class="module-alert alert alert-warning">
+            <h3>{{ _("The Recombinant resource is out of date") }}{% snippet 'snippets/sysadmin_only.html' %}</h3>
+            <p style="margin-bottom: 10px;">{{ _('You can refresh your resource in the database to try to solve the problem.') }}</p>
+            <form id="create-pd-resource" method="post">
+              {{ h.csrf_input() }}
+              <button type="submit" class="btn btn-danger mrgn-bttm-md m-b-3" name="refresh-hard">{{_('Refresh…')}}</button>
+            </form>
+          </div>
+        {% endif %}
         <div class="wb-tabs">
           <div class="tabpanels">
             {% block update_panel %}
@@ -59,6 +70,34 @@
                   {% snippet "recombinant/snippets/xls_download.html",
                     pkg=dataset, errors=errors %}
                 {% endif %}
+                {% if g.userobj.sysadmin and resource.datastore_active %}
+                  {# only let sysadmins datastore dump via UI #}
+                  {% set filename = dataset.owner_org ~ '.' ~ resource.name %}
+                  <h3>{{ _('DataStore Dump') }}{% snippet 'snippets/sysadmin_only.html' %}</h3>
+                  <div>
+                    <button class="btn btn-primary dropdown-toggle" role="button" id="dropdownDownloadFormat" data-bs-toggle="dropdown" aria-expanded="false" aria-label="{{ _('List of downloadable formats') }}">
+                        {{ _('Download') }}&nbsp;
+                    </button>
+                    <ul class="dropdown-menu" aria-labelledby="dropdownDownloadFormat">
+                      <li>
+                        <a class="dropdown-item" href="{{ h.url_for('datastore.dump', resource_id=resource.id, bom=True, filename=filename) }}"
+                          target="_blank" rel="noreferrer"><span>CSV</span></a>
+                      </li>
+                      <li>
+                        <a class="dropdown-item" href="{{ h.url_for('datastore.dump', resource_id=resource.id, format='tsv', bom=True, filename=filename) }}"
+                          target="_blank" rel="noreferrer"><span>TSV</span></a>
+                      </li>
+                      <li>
+                          <a class="dropdown-item" href="{{ h.url_for('datastore.dump', resource_id=resource.id, format='json', filename=filename) }}"
+                          target="_blank" rel="noreferrer"><span>JSON</span></a>
+                      </li>
+                      <li>
+                          <a class="dropdown-item" href="{{ h.url_for('datastore.dump', resource_id=resource.id, format='xml', filename=filename) }}"
+                          target="_blank" rel="noreferrer"><span>XML</span></a>
+                      </li>
+                    </ul>
+                  </div>
+                {% endif %}
                 {% block notices %}{% endblock %}
               </details>
             {% endblock %}
@@ -108,6 +147,7 @@ <h3>{{_("Error loading your records")}}</h3>
                 <p>{{_("We were unable to retrieve your proactive publication records from our database. Please contact <a href=\"open-ouvert@tbs-sct.gc.ca\">open-ouvert@tbs-sct.gc.ca</a>.")}}</p>
                 {% if g.userobj.sysadmin %}
                   {# only let sysadmins refresh the recombinant record via UI #}
+                  <h4>{{ _('Refresh resource') }}{% snippet 'snippets/sysadmin_only.html' %}</h4>
                   <form id="create-pd-resource" method="post">
                     {{ h.csrf_input() }}
                     <button type="submit" class="btn btn-default mrgn-bttm-md m-b-3" name="refresh">{{_('Refresh…')}}</button>

ckanext/recombinant/templates/snippets/sysadmin_only.html

@@ -0,0 +1 @@
+&nbsp;<sup style="cursor: help;" title="{{ _('Sysadmin only') }}"><small><span class="fa fa-gavel text-info" aria-hidden="true"></span></small></sup>

ckanext/recombinant/views.py

@@ -26,7 +26,7 @@
 
 from ckan.logic import ValidationError, NotAuthorized
 from ckan.model.group import Group
-from ckan.authz import has_user_permission_for_group_or_org
+from ckan.authz import has_user_permission_for_group_or_org, is_sysadmin
 
 from ckan.views.dataset import _get_package_type
 
@@ -558,7 +558,10 @@ def preview_table(resource_name: str,
     except RecombinantException:
         return abort(404, _('Recombinant resource_name not found'))
 
-    if 'create' in request.form or 'refresh' in request.form:
+    if (
+      'create' in request.form or
+      'refresh-hard' in request.form or
+      'refresh' in request.form):
         # check if the user can update datasets for organization
         # admin and editors should be able to init recombinant records
         if not has_user_permission_for_group_or_org(org_object.id,
@@ -572,17 +575,26 @@ def preview_table(resource_name: str,
                 dataset_type=chromo['dataset_type'], owner_org=owner_org)
             # check that the resource has errors
             for _r in dataset['resources']:
-                if _r['name'] == resource_name and 'error' in _r:
+                if _r['name'] == resource_name and ('error' in _r or
+                                                    not _r['datastore_correct']):
                     raise NotFound
         except NotFound:
             try:
                 if 'create' in request.form:
                     lc.action.recombinant_create(
                         dataset_type=chromo['dataset_type'], owner_org=owner_org)
-                else:
+                elif 'refresh-hard' in request.form or 'refresh' in request.form:
+                    if not is_sysadmin(g.user):
+                        # only sysadmins can refresh via UI
+                        return abort(403)
+                    delete_fields = False
+                    if 'refresh-hard' in request.form:
+                        delete_fields = True
                     lc.action.recombinant_update(
                         dataset_type=chromo['dataset_type'], owner_org=owner_org,
-                        force_update=True)
+                        force_update=True,
+                        delete_fields=delete_fields)
+                    h.flash_success(_('Resources successfully refreshed.'))
             except NotAuthorized as e:
                 return abort(403, e.message or '')
         return h.redirect_to(