Clean Datastore Tables Job

ckanext/xloader/config_declaration.yaml

@@ -112,5 +112,13 @@ groups:
             to True.
         type: bool
         required: false
+      - key: ckanext.xloader.clean_datastore_tables
+        default: False
+        example: True
+        description: |
+          Enqueue jobs to remove Datastore tables from Resources that have a format
+          that is not in ckanext.xloader.formats after a Resource is updated.
+        type: bool
+        required: false
 
 

ckanext/xloader/plugin.py

@@ -4,6 +4,8 @@
 
 from ckan import plugins
 from ckan.plugins import toolkit
+from ckan.model.domain_object import DomainObjectOperation
+from ckan.model.resource import Resource
 
 from . import action, auth, helpers as xloader_helpers, utils
 from .loader import fulltext_function_exists, get_write_engine
@@ -53,7 +55,7 @@ def is_it_an_xloader_format(cls, format_):
 class xloaderPlugin(plugins.SingletonPlugin):
     plugins.implements(plugins.IConfigurer)
     plugins.implements(plugins.IConfigurable)
-    plugins.implements(plugins.IResourceUrlChange)
+    plugins.implements(plugins.IDomainObjectModification)
     plugins.implements(plugins.IActions)
     plugins.implements(plugins.IAuthFunctions)
     plugins.implements(plugins.ITemplateHelpers)
@@ -94,18 +96,32 @@ def configure(self, config_):
                     )
                 )
 
-    # IResourceUrlChange
+    # IDomainObjectModification
+
+    def notify(self, entity, operation):
+        # type: (ckan.model.Package|ckan.model.Resource, DomainObjectOperation) -> None
+        """
+        Runs before_commit to database for Packages and Resources.
+        We only want to check for changed Resources for this.
+        We want to check if values have changed, namely the url and the format.
+        See: ckan/model/modification.py.DomainObjectModificationExtension
+        """
+        if operation != DomainObjectOperation.changed or not isinstance(entity, Resource):
+            return
 
-    def notify(self, resource):
         context = {
             "ignore_auth": True,
         }
         resource_dict = toolkit.get_action("resource_show")(
             context,
             {
-                "id": resource.id,
+                "id": entity.id,
             },
         )
+
+        if _should_remove_unsupported_resource_from_datastore(resource_dict):
+            toolkit.enqueue_job(fn=_remove_unsupported_resource_from_datastore, args=[entity.id])
+
         self._submit_to_xloader(resource_dict)
 
     # IResourceController
@@ -212,3 +228,39 @@ def get_helpers(self):
             "xloader_status": xloader_helpers.xloader_status,
             "xloader_status_description": xloader_helpers.xloader_status_description,
         }
+
+
+def _should_remove_unsupported_resource_from_datastore(res_dict):
+    if not toolkit.asbool(toolkit.config.get('ckanext.xloader.clean_datastore_tables', False)):
+        return False
+    return (not XLoaderFormats.is_it_an_xloader_format(res_dict.get('format', u''))
+            and (res_dict.get('url_type') == 'upload'
+                or res_dict.get('url_type') == '')
+            and (res_dict.get('datastore_active', False)
+                or res_dict.get('extras', {}).get('datastore_active', False)))
+
+
+def _remove_unsupported_resource_from_datastore(resource_id):
+    """
+    Callback to remove unsupported datastore tables.
+    Controlled by config value: ckanext.xloader.clean_datastore_tables.
+    Double check the resource format. Only supported Xloader formats should have datastore tables.
+    If the resource format is not supported, we should delete the datastore tables.
+    """
+    context = {"ignore_auth": True}
+    try:
+        res = toolkit.get_action('resource_show')(context, {"id": resource_id})
+    except toolkit.ObjectNotFound:
+        log.error('Resource %s does not exist.' % res['id'])
+        return
+
+    if _should_remove_unsupported_resource_from_datastore(res):
+        log.info('Unsupported resource format "%s". Deleting datastore tables for resource %s'
+            % (res.get(u'format', u'').lower(), res['id']))
+        try:
+            toolkit.get_action('datastore_delete')(context, {
+                "resource_id": res['id'],
+                "force": True})
+            log.info('Datastore table dropped for resource %s' % res['id'])
+        except toolkit.ObjectNotFound:
+            log.error('Datastore table for resource %s does not exist' % res['id'])