Validation Extension Support

ckanext/xloader/action.py

@@ -60,7 +60,6 @@ def xloader_submit(context, data_dict):
     p.toolkit.check_access('xloader_submit', context, data_dict)
 
     sync = data_dict.pop('sync', False)
-
     res_id = data_dict['resource_id']
     try:
         resource_dict = p.toolkit.get_action('resource_show')(context, {

ckanext/xloader/config_declaration.yaml

@@ -138,14 +138,6 @@ groups:
 
             See https://github.com/frictionlessdata/ckanext-validation?tab=readme-ov-file#data-schema
             for more details.
-      - key: ckanext.xloader.validation.chain_xloader
-        default: True
-        example: False
-        description: |
-            Resources that pass Validation will immediately get XLoadered instead of having
-            a job enqueued for it.
-
-            If this option is set to `False`, jobs will be enqueued like normal.
       - key: ckanext.xloader.clean_datastore_tables
         default: False
         example: True

ckanext/xloader/plugin.py

@@ -12,6 +12,12 @@
 from . import action, auth, helpers as xloader_helpers, utils
 from ckanext.xloader.utils import XLoaderFormats
 
+try:
+    from ckanext.validation.interfaces import IPipeValidation
+    HAS_IPIPE_VALIDATION = True
+except ImportError:
+    HAS_IPIPE_VALIDATION = False
+
 try:
     config_declarations = toolkit.blanket.config_declarations
 except AttributeError:
@@ -34,6 +40,8 @@ class xloaderPlugin(plugins.SingletonPlugin):
     plugins.implements(plugins.IResourceController, inherit=True)
     plugins.implements(plugins.IClick)
     plugins.implements(plugins.IBlueprint)
+    if HAS_IPIPE_VALIDATION:
+        plugins.implements(IPipeValidation)
 
     # IClick
     def get_commands(self):
@@ -68,6 +76,23 @@ def configure(self, config_):
                     )
                 )
 
+    # IPipeValidation
+
+    def receive_validation_report(self, validation_report):
+        if utils.requires_successful_validation_report():
+            res_dict = toolkit.get_action('resource_show')({'ignore_auth': True},
+                                                           {'id': validation_report.get('resource_id')})
+            if (toolkit.asbool(toolkit.config.get('ckanext.xloader.validation.enforce_schema', True))
+                or res_dict.get('schema', None)) and validation_report.get('status') != 'success':
+                    # either validation.enforce_schema is turned on or it is off and there is a schema,
+                    # we then explicitly check for the `validation_status` report to be `success`
+                    return
+            # if validation is running in async mode, it is running from the redis workers.
+            # thus we need to do sync=True to have Xloader put the job at the front of the queue.
+            sync = toolkit.asbool(toolkit.config.get(u'ckanext.validation.run_on_update_async', True))
+            self._submit_to_xloader(res_dict, sync=sync)
+
+
     # IDomainObjectModification
 
     def notify(self, entity, operation):
@@ -95,21 +120,10 @@ def notify(self, entity, operation):
         if _should_remove_unsupported_resource_from_datastore(resource_dict):
             toolkit.enqueue_job(fn=_remove_unsupported_resource_from_datastore, args=[entity.id])
 
-        if utils.awaiting_validation(resource_dict):
-            # If the resource requires validation, stop here if validation
-            # has not been performed or did not succeed. The Validation
-            # extension will call resource_patch and this method should
-            # be called again. However, url_changed will not be in the entity
-            # once Validation does the patch.
+        if utils.requires_successful_validation_report():
             log.debug("Skipping xloading resource %s because the "
                       "resource did not pass validation yet.", entity.id)
             return
-        elif utils.do_chain_after_validation(resource_dict):
-            # At this point, the Resource has passed validation requirements,
-            # and chaining is turned on. We will execute XLoader right away,
-            # inside of the Validation job, instead of enqueueing a job.
-            self._submit_to_xloader(resource_dict, sync=True)
-            return
         elif not getattr(entity, 'url_changed', False):
             # do not submit to xloader if the url has not changed.
             return
@@ -119,15 +133,11 @@ def notify(self, entity, operation):
     # IResourceController
 
     def after_resource_create(self, context, resource_dict):
-        if utils.awaiting_validation(resource_dict):
+        if utils.requires_successful_validation_report():
             log.debug("Skipping xloading resource %s because the "
                       "resource did not pass validation yet.", resource_dict.get('id'))
             return
 
-        if utils.do_chain_after_validation(resource_dict):
-            self._submit_to_xloader(resource_dict, sync=True)
-            return
-
         self._submit_to_xloader(resource_dict)
 
     def before_resource_show(self, resource_dict):

ckanext/xloader/utils.py

@@ -48,6 +48,10 @@ def is_it_an_xloader_format(cls, format_):
         return format_.lower() in cls._formats
 
 
+def requires_successful_validation_report():
+    return p.toolkit.asbool(config.get('ckanext.xloader.validation.requires_successful_report', False))
+
+
 def awaiting_validation(res_dict):
     # type: (dict) -> bool
     """
@@ -60,7 +64,7 @@ def awaiting_validation(res_dict):
     Checks ckanext.xloader.validation.enforce_schema config
     option value. Then checks the Resource's validation_status.
     """
-    if not p.toolkit.asbool(config.get('ckanext.xloader.validation.requires_successful_report', False)):
+    if not requires_successful_validation_report():
         # validation.requires_successful_report is turned off, return right away
         return False
 
@@ -90,37 +94,6 @@ def awaiting_validation(res_dict):
     return False
 
 
-def do_chain_after_validation(res_dict):
-    # type: (dict) -> bool
-
-    current_job = get_current_job()
-
-    if not p.toolkit.asbool(config.get('ckanext.xloader.validation.requires_successful_report', False)) \
-            or not p.toolkit.asbool(config.get('ckanext.xloader.validation.chain_xloader', True)) \
-            or not current_job:
-
-        # we are not requiring resources to pass validation
-        # OR we are not chaining validation to xloader
-        # OR we are outside of the job context, thus not running a job
-        return False
-
-    if current_job.func_name != 'ckanext.validation.jobs.run_validation_job':
-        # the current running job is not the ckanext-validation validate job
-        #FIXME: how to do a better check for the caller in the stack??
-        return False
-
-    try:
-        job_rid = current_job.args[0].get('id', None)
-    except (KeyError):
-        job_rid = None
-    if res_dict.get('id', None) != job_rid:
-        # the current running job's Resource ID is not
-        # the same as the passed Resource's ID
-        return False
-
-    return True
-
-
 def resource_data(id, resource_id, rows=None):
 
     if p.toolkit.request.method == "POST":