Frontend Status Badges

.github/workflows/test.yml

@@ -53,6 +53,8 @@ jobs:
 
     steps:
     - uses: actions/checkout@v4
+    - if: ${{ matrix.ckan-version == 2.9 }}
+      run: pip install "setuptools>=44.1.0,<71"
     - name: Install requirements
       run: |
         pip install -r requirements.txt

README.rst

@@ -128,7 +128,7 @@ 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.9             yes (Python3) (last supported version for Python 2.7: 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
 =============== =============

ckanext/xloader/config_declaration.yaml

@@ -128,6 +128,23 @@ groups:
             to True.
         type: bool
         required: false
+      - key: ckanext.xloader.validation.requires_successful_report
+        default: False
+        example: True
+        description: |
+            Resources are required to pass Validation from the ckanext-validation
+            plugin to be able to get XLoadered.
+        type: bool
+        required: false
+      - key: ckanext.xloader.validation.enforce_schema
+        default: True
+        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
         default: False
         example: True
@@ -136,5 +153,21 @@ groups:
           that is not in ckanext.xloader.formats after a Resource is updated.
         type: bool
         required: false
+      - key: ckanext.xloader.show_badges
+        default: True
+        example: False
+        description: |
+          Controls whether or not the status badges display in the front end.
+        type: bool
+        required: false
+      - key: ckanext.xloader.debug_badges
+        default: False
+        example: True
+        description: |
+          Controls whether or not the status badges display all of the statuses. By default,
+          the badges will display "pending", "running", and "error". With debug_badges enabled,
+          they will also display "complete", "active", "inactive", and "unknown".
+        type: bool
+        required: false
 
 

ckanext/xloader/helpers.py

@@ -1,5 +1,7 @@
 import ckan.plugins.toolkit as toolkit
 from ckanext.xloader.utils import XLoaderFormats
+from markupsafe import Markup
+from html import escape as html_escape
 
 
 def xloader_status(resource_id):
@@ -42,3 +44,82 @@ def is_resource_supported_by_xloader(res_dict, check_access=True):
     else:
         is_supported_url_type = True
     return (is_supported_format or is_datastore_active) and user_has_access and is_supported_url_type
+
+
+def xloader_badge(resource):
+    # type: (dict) -> str
+    """
+    Displays a custom badge for the status of Xloader and DataStore for the specified resource.
+    """
+    if not toolkit.asbool(toolkit.config.get('ckanext.xloader.show_badges', True)):
+        return ''
+
+    if not XLoaderFormats.is_it_an_xloader_format(resource.get('format')):
+        # we only want to show badges for supported xloader formats
+        return ''
+
+    is_datastore_active = resource.get('datastore_active', False)
+
+    try:
+        xloader_job = toolkit.get_action("xloader_status")({'ignore_auth': True},
+                                                           {"resource_id": resource.get('id')})
+    except toolkit.ObjectNotFound:
+        xloader_job = {}
+
+    if xloader_job.get('status') == 'complete':
+        # the xloader task is complete, show datastore active or inactive.
+        # xloader will delete the datastore table at the beggining of the job run.
+        # so this will only be true if the job is fully finished.
+        status = 'active' if is_datastore_active else 'inactive'
+    elif xloader_job.get('status') in ['submitting', 'pending', 'running', 'running_but_viewable', 'error']:
+        # the job is running or pending or errored
+        # show the xloader status
+        status = xloader_job.get('status')
+        if status == 'running_but_viewable':
+            # treat running_but_viewable the same as running
+            status = 'running'
+        elif status == 'submitting':
+            # treat submitting the same as pending
+            status = 'pending'
+    else:
+        # we do not know what the status is
+        status = 'unknown'
+
+    messages = {
+        # Default messages
+        'pending': toolkit._('Data awaiting load to DataStore'),
+        'running': toolkit._('Loading data into DataStore'),
+        'error': toolkit._('Failed to load data into DataStore'),
+        # Debug messages
+        'complete': toolkit._('Data loaded into DataStore'),
+        'active': toolkit._('Data available in DataStore'),
+        'inactive': toolkit._('Resource not active in DataStore'),
+        'unknown': toolkit._('DataStore status unknown'),
+    }
+    basic_statuses = ['pending', 'running', 'error']
+
+    if status not in basic_statuses and not toolkit.asbool(toolkit.config.get('ckanext.xloader.debug_badges', False)):
+        return ''
+
+    badge_url = toolkit.h.url_for_static('/static/badges/{lang}/datastore-{status}.svg'.format(
+        lang=toolkit.h.lang(), status=status))
+
+    title = toolkit.h.render_datetime(xloader_job.get('last_updated'), with_hours=True) \
+        if xloader_job.get('last_updated') else ''
+
+    try:
+        toolkit.check_access('resource_update', {'user': toolkit.g.user}, {'id': resource.get('id')})
+        pusher_url = toolkit.h.url_for('xloader.resource_data',
+                                       id=resource.get('package_id'),
+                                       resource_id=resource.get('id'))
+
+        return Markup(u'<a href="{pusher_url}" class="loader-badge"><img src="{badge_url}" alt="{alt}" title="{title}"/></a>'.format(
+            pusher_url=pusher_url,
+            badge_url=badge_url,
+            alt=html_escape(messages[status], quote=True),
+            title=html_escape(title, quote=True)))
+    except toolkit.NotAuthorized:
+        return Markup(u'<span class="loader-badge"><img src="{badge_url}" alt="{alt}" title="{title}"/></span>'.format(
+            badge_url=badge_url,
+            alt=html_escape(messages[status], quote=True),
+            title=html_escape(title, quote=True)))

ckanext/xloader/jobs.py

@@ -124,6 +124,7 @@ def xloader_data_into_datastore(input):
             if tries < MAX_RETRIES:
                 tries = tries + 1
                 log.info("Job %s failed due to temporary error [%s], retrying", job_id, e)
+                logger.info("Job failed due to temporary error [%s], retrying", e)
                 job_dict['status'] = 'pending'
                 job_dict['metadata']['tries'] = tries
                 enqueue_job(
@@ -245,7 +246,12 @@ def tabulator_load():
     logger.info("'use_type_guessing' mode is: %s", use_type_guessing)
     try:
         if use_type_guessing:
-            tabulator_load()
+            try:
+                tabulator_load()
+            except JobError as e:
+                logger.warning('Load using tabulator failed: %s', e)
+                logger.info('Trying again with direct COPY')
+                direct_load()
         else:
             try:
                 direct_load()

ckanext/xloader/loader.py

@@ -118,8 +118,8 @@ def _clear_datastore_resource(resource_id):
     '''
     engine = get_write_engine()
     with engine.begin() as conn:
-        conn.execute("SET LOCAL lock_timeout = '5s'")
-        conn.execute('TRUNCATE TABLE "{}"'.format(resource_id))
+        conn.execute("SET LOCAL lock_timeout = '15s'")
+        conn.execute('TRUNCATE TABLE "{}" RESTART IDENTITY'.format(resource_id))
 
 
 def load_csv(csv_filepath, resource_id, mimetype='text/csv', logger=None):
@@ -339,6 +339,18 @@ def create_column_indexes(fields, resource_id, logger):
     logger.info('...column indexes created.')
 
 
+def _save_type_overrides(headers_dicts):
+    # copy 'type' to 'type_override' if it's not the default type (text)
+    # and there isn't already an override in place
+    for h in headers_dicts:
+        if h['type'] != 'text':
+            if 'info' in h:
+                if 'type_override' not in h['info']:
+                    h['info']['type_override'] = h['type']
+            else:
+                h['info'] = {'type_override': h['type']}
+
+
 def load_table(table_filepath, resource_id, mimetype='text/csv', logger=None):
     '''Loads an Excel file (or other tabular data recognized by tabulator)
     into Datastore and creates indexes.
@@ -399,7 +411,14 @@ def load_table(table_filepath, resource_id, mimetype='text/csv', logger=None):
             }.get(existing_info.get(h, {}).get('type_override'), t)
             for t, h in zip(types, headers)]
 
-    headers = [header.strip()[:MAX_COLUMN_LENGTH] for header in headers if header.strip()]
+    # Strip leading and trailing whitespace, then truncate to maximum length,
+    # then strip again in case the truncation exposed a space.
+    headers = [
+        header.strip()[:MAX_COLUMN_LENGTH].strip()
+        for header in headers
+        if header and header.strip()
+    ]
+    header_count = len(headers)
     type_converter = TypeConverter(types=types)
 
     with UnknownEncodingStream(table_filepath, file_format, decoding_result,
@@ -409,6 +428,17 @@ def row_iterator():
             for row in stream:
                 data_row = {}
                 for index, cell in enumerate(row):
+                    # Handle files that have extra blank cells in heading and body
+                    # eg from Microsoft Excel adding lots of empty cells on export.
+                    # Blank header cells won't generate a column,
+                    # so row length won't match column count.
+                    if index >= header_count:
+                        # error if there's actual data out of bounds, otherwise ignore
+                        if cell:
+                            raise LoaderError("Found data in column %s but resource only has %s header(s)",
+                                              index + 1, header_count)
+                        else:
+                            continue
                     data_row[headers[index]] = cell
                 yield data_row
         result = row_iterator()
@@ -426,6 +456,9 @@ def row_iterator():
                     if type_override in list(_TYPE_MAPPING.values()):
                         h['type'] = type_override
 
+        # preserve any types that we have sniffed unless told otherwise
+        _save_type_overrides(headers_dicts)
+
         logger.info('Determined headers and types: %s', headers_dicts)
 
         '''

ckanext/xloader/plugin.py

@@ -7,11 +7,16 @@
 
 from ckan.model.domain_object import DomainObjectOperation
 from ckan.model.resource import Resource
-from ckan.model.package import Package
 
 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 +39,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):
@@ -51,6 +58,7 @@ def get_blueprint(self):
 
     def update_config(self, config):
         toolkit.add_template_directory(config, 'templates')
+        toolkit.add_public_directory(config, 'public')
 
     # IConfigurable
 
@@ -68,6 +76,21 @@ 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':
+                    # A schema is present, or required to be present
+                    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,7 +118,16 @@ 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 not getattr(entity, 'url_changed', False):
+        if utils.requires_successful_validation_report():
+            # 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.
+            log.debug("Deferring xloading resource %s because the "
+                      "resource did not pass validation yet.", resource_dict.get('id'))
+            return
+        elif not getattr(entity, 'url_changed', False):
             # do not submit to xloader if the url has not changed.
             return
 
@@ -104,6 +136,11 @@ def notify(self, entity, operation):
     # IResourceController
 
     def after_resource_create(self, context, resource_dict):
+        if utils.requires_successful_validation_report():
+            log.debug("Deferring xloading resource %s because the "
+                      "resource did not pass validation yet.", resource_dict.get('id'))
+            return
+
         self._submit_to_xloader(resource_dict)
 
     def before_resource_show(self, resource_dict):
@@ -206,6 +243,7 @@ def get_helpers(self):
             "xloader_status": xloader_helpers.xloader_status,
             "xloader_status_description": xloader_helpers.xloader_status_description,
             "is_resource_supported_by_xloader": xloader_helpers.is_resource_supported_by_xloader,
+            "xloader_badge": xloader_helpers.xloader_badge,
         }