Validation Extension Support

CHANGELOG

@@ -1,3 +1,12 @@
+1.1.3 2024-12-09
+================
+
+Feat:
+* Adds support for ckanext-validation. Config `ckanext.xloader.validation.requires_successful_report` controls whether a resource requires a successful validation report to be XLoadered. By default, a resource would also require a Validation Schema, which can be turned off with `ckanext.xloader.validation.enforce_schema`.
+
+
+**Full Changelog**: https://github.com/ckan/ckanext-xloader/compare/1.1.2...1.1.3
+
 1.1.2 2024-10-25
 ================
 
@@ -21,7 +30,7 @@ Fix:
 1.1.0 2024-10-16
 ================
 
-Fixes: 
+Fixes:
 * feat: Add pypi cicd publish via github action via environment controls by @duttonw in https://github.com/ckan/ckanext-xloader/pull/228
 
 
@@ -32,7 +41,7 @@ Fixes:
 ================
 
 
-Fixes: 
+Fixes:
 
 * add README note about running on separate server, #191 by @ThrawnCA in https://github.com/ckan/ckanext-xloader/pull/192
 * Use IDomainObjectModification Implementation by @JVickery-TBS in https://github.com/ckan/ckanext-xloader/pull/198

README.md

@@ -115,7 +115,7 @@ Compatibility with core CKAN versions:
   | CKAN version   | Compatibility                                                                                                  |
   | -------------- | -------------------------------------------------------------------------------------------------------------- |
   | 2.7            | no longer supported (last supported version: 0.12.2)                                                           |
-  | 2.8            | no longer supported (last supported version: 0.12.2)                                                           | 
+  | 2.8            | no longer supported (last supported version: 0.12.2)                                                           |
   | 2.9            | yes (Python3) (last supported version for Python 2.7: 0.12.2)), Must: `pip install "setuptools>=44.1.0,<71"`   |
   | 2.10           | yes                                                                                                            |
   | 2.11           | yes                                                                                                            |
@@ -189,7 +189,7 @@ expect European (day-first) dates, you could add to `postgresql.conf`:
 
     datestyle=ISO,DMY
 
-All configurations below are defined in the 
+All configurations below are defined in the
 [config_declaration.yaml](ckanext/xloader/config_declaration.yaml) file.
 
 
@@ -409,6 +409,34 @@ Controls whether or not the status badges display all of the statuses. By defaul
 the badges will display "pending", "running", and "error". With debug_badges enabled,
 they will also display "complete", "active", "inactive", and "unknown".
 
+#### ckanext.xloader.validation.requires_successful_report
+
+Supports: __ckanext-validation__
+
+Example:
+
+```
+ckanext.xloader.validation.requires_successful_report = True
+```
+
+Default value: `False`
+
+Controls whether or not a resource requires a successful validation report from the ckanext-validation plugin in order to be XLoadered.
+
+#### ckanext.xloader.validation.enforce_schema
+
+Supports: __ckanext-validation__
+
+Example:
+
+```
+ckanext.xloader.validation.enforce_schema = False
+```
+
+Default value: `True`
+
+Controls whether or not a resource requires a Validation Schema to be present from the ckanext-validation plugin to be XLoadered.
+
 ## Developer installation
 
 To install XLoader for development, activate your CKAN virtualenv and in

ckanext/xloader/action.py

@@ -53,6 +53,13 @@ def xloader_submit(context, data_dict):
     if errors:
         raise p.toolkit.ValidationError(errors)
 
+    p.toolkit.check_access('xloader_submit', context, data_dict)
+
+    # If sync is set to True, the xloader callback will be executed right
+    # away, instead of a job being enqueued. It will also delete any existing jobs
+    # for the given resource. This is only controlled by sysadmins or the system.
+    sync = data_dict.pop('sync', False)
+
     res_id = data_dict['resource_id']
     try:
         resource_dict = p.toolkit.get_action('resource_show')(context, {
@@ -166,15 +173,20 @@ def xloader_submit(context, data_dict):
         job = enqueue_job(
             jobs.xloader_data_into_datastore, [data], queue=custom_queue,
             title="xloader_submit: package: {} resource: {}".format(resource_dict.get('package_id'), res_id),
-            rq_kwargs=dict(timeout=timeout)
+            rq_kwargs=dict(timeout=timeout, at_front=sync)
         )
     except Exception:
-        log.exception('Unable to enqueued xloader res_id=%s', res_id)
+        if sync:
+            log.exception('Unable to xloader res_id=%s', res_id)
+        else:
+            log.exception('Unable to enqueue xloader res_id=%s', res_id)
         return False
     log.debug('Enqueued xloader job=%s res_id=%s', job.id, res_id)
-
     value = json.dumps({'job_id': job.id})
 
+    if sync:
+        log.debug('Pushed xloader sync mode job=%s res_id=%s to front of queue', job.id, res_id)
+
     task['value'] = value
     task['state'] = 'pending'
     task['last_updated'] = str(datetime.datetime.utcnow())

ckanext/xloader/config_declaration.yaml

@@ -141,8 +141,10 @@ groups:
         example: False
         description: |
             Resources are expected to have a Validation Schema, or use the default ones if not.
+
             If this option is set to `False`, Resources that do not have
             a Validation Schema will be treated like they do not require Validation.
+
             See https://github.com/frictionlessdata/ckanext-validation?tab=readme-ov-file#data-schema
             for more details.
       - key: ckanext.xloader.clean_datastore_tables

ckanext/xloader/plugin.py

@@ -183,7 +183,7 @@ def before_show(self, resource_dict):
         def after_update(self, context, resource_dict):
             self.after_resource_update(context, resource_dict)
 
-    def _submit_to_xloader(self, resource_dict):
+    def _submit_to_xloader(self, resource_dict, sync=False):
         context = {"ignore_auth": True, "defer_commit": True}
         resource_format = resource_dict.get("format")
         if not XLoaderFormats.is_it_an_xloader_format(resource_format):
@@ -203,14 +203,20 @@ def _submit_to_xloader(self, resource_dict):
             return
 
         try:
-            log.debug(
-                "Submitting resource %s to be xloadered", resource_dict["id"]
-            )
+            if sync:
+                log.debug(
+                    "xloadering resource %s in sync mode", resource_dict["id"]
+                )
+            else:
+                log.debug(
+                    "Submitting resource %s to be xloadered", resource_dict["id"]
+                )
             toolkit.get_action("xloader_submit")(
                 context,
                 {
                     "resource_id": resource_dict["id"],
                     "ignore_hash": self.ignore_hash,
+                    "sync": sync,
                 },
             )
         except toolkit.ValidationError as e:

ckanext/xloader/schema.py

@@ -16,6 +16,7 @@
 boolean_validator = get_validator('boolean_validator')
 int_validator = get_validator('int_validator')
 OneOf = get_validator('OneOf')
+ignore_not_sysadmin = get_validator('ignore_not_sysadmin')
 
 if p.toolkit.check_ckan_version('2.9'):
     unicode_safe = get_validator('unicode_safe')
@@ -29,6 +30,7 @@ def xloader_submit_schema():
         'id': [ignore_missing],
         'set_url_type': [ignore_missing, boolean_validator],
         'ignore_hash': [ignore_missing, boolean_validator],
+        'sync': [ignore_missing, boolean_validator, ignore_not_sysadmin],
         '__junk': [empty],
         '__before': [dsschema.rename('id', 'resource_id')]
     }

ckanext/xloader/utils.py

@@ -2,6 +2,7 @@
 
 import json
 import datetime
+from rq import get_current_job
 
 from six import text_type as str, binary_type
 
@@ -33,8 +34,6 @@
     "application/vnd.oasis.opendocument.spreadsheet",
 ]
 
-from .job_exceptions import JobError
-
 
 class XLoaderFormats(object):
     formats = None
@@ -61,8 +60,10 @@ def awaiting_validation(res_dict):
     """
     Checks the existence of a logic action from the ckanext-validation
     plugin, thus supporting any extending of the Validation Plugin class.
+
     Checks ckanext.xloader.validation.requires_successful_report config
     option value.
+
     Checks ckanext.xloader.validation.enforce_schema config
     option value. Then checks the Resource's validation_status.
     """
@@ -273,7 +274,7 @@ def type_guess(rows, types=TYPES, strict=False):
         at_least_one_value = []
         for ri, row in enumerate(rows):
             diff = len(row) - len(guesses)
-            for _ in range(diff):
+            for _i in range(diff):
                 typesdict = {}
                 for type in types:
                     typesdict[type] = 0