feat(dev): IPipeValidation interface;

- Created new `IPipeValidation` interface which sends dictized validation reports to other plugins.

ckanext/validation/interfaces.py

@@ -39,3 +39,21 @@ def can_validate(self, context, data_dict):
 
         '''
         return True
+
+
+class IPipeValidation(Interface):
+    """
+    Process data in a Data Pipeline.
+
+    Inherit this to subscribe to events in the Data Pipeline and be able to
+    broadcast the results for others to process next. In this way, a number of
+    IPipes can be linked up in sequence to build up a data processing pipeline.
+
+    When a resource is validated, it broadcasts its validation_report,
+    perhaps triggering a process which transforms the data to another format,
+    or loads it into a datastore. These processes can in turn put the resulting
+    validation reports into the pipeline
+    """
+
+    def receive_validation_report(self, validation_report):
+        pass

ckanext/validation/jobs.py

@@ -15,7 +15,11 @@
 import ckantoolkit as t
 
 from ckanext.validation.model import Validation
-from ckanext.validation.utils import get_update_mode_from_config
+from ckanext.validation.utils import (
+    get_update_mode_from_config,
+    send_validation_report,
+    validation_dictize,
+)
 
 
 log = logging.getLogger(__name__)
@@ -132,6 +136,7 @@ def run_validation_job(resource):
         '_validation_performed': True
     }
     t.get_action('resource_patch')(patch_context, data_dict)
+    send_validation_report(validation_dictize(validation))
 
 
 

ckanext/validation/logic.py

@@ -19,6 +19,7 @@
     get_create_mode_from_config,
     get_update_mode_from_config,
     delete_local_uploaded_file,
+    validation_dictize,
 )
 
 
@@ -172,7 +173,7 @@ def resource_validation_show(context, data_dict):
         raise t.ObjectNotFound(
             'No validation report exists for this resource')
 
-    return _validation_dictize(validation)
+    return validation_dictize(validation)
 
 
 def resource_validation_delete(context, data_dict):
@@ -404,22 +405,6 @@ def _add_default_formats(search_data_dict):
     search_data_dict['fq_list'].append(' OR '.join(filter_formats_query))
 
 
-def _validation_dictize(validation):
-    out = {
-        'id': validation.id,
-        'resource_id': validation.resource_id,
-        'status': validation.status,
-        'report': validation.report,
-        'error': validation.error,
-    }
-    out['created'] = (
-        validation.created.isoformat() if validation.created else None)
-    out['finished'] = (
-        validation.finished.isoformat() if validation.finished else None)
-
-    return out
-
-
 @t.chained_action
 def resource_create(up_func, context, data_dict):
     '''Appends a new resource to a datasets list of resources.

ckanext/validation/utils.py

@@ -1,9 +1,12 @@
 import os
 import logging
 
+from ckan.plugins import PluginImplementations
 from ckan.lib.uploader import ResourceUpload
 from ckantoolkit import config, asbool
 
+from ckanext.validation.interfaces import IPipeValidation
+
 
 log = logging.getLogger(__name__)
 
@@ -71,3 +74,30 @@ def delete_local_uploaded_file(resource_id):
 
     except OSError as e:
         log.warning(u'Error deleting uploaded file: %s', e)
+
+
+def validation_dictize(validation):
+    out = {
+        'id': validation.id,
+        'resource_id': validation.resource_id,
+        'status': validation.status,
+        'report': validation.report,
+        'error': validation.error,
+    }
+    out['created'] = (
+        validation.created.isoformat() if validation.created else None)
+    out['finished'] = (
+        validation.finished.isoformat() if validation.finished else None)
+
+    return out
+
+
+def send_validation_report(validation_report):
+    for observer in PluginImplementations(IPipeValidation):
+        try:
+            observer.receive_validation_report(validation_report)
+        except Exception as ex:
+            log.exception(ex)
+            # We reraise all exceptions so they are obvious there
+            # is something wrong
+            raise