feat(dev): IPipeXloader interface;

- Made `IPipeXloader` interface to send xloadr status to other plugins.

ckanext/xloader/action.py

@@ -311,6 +311,8 @@ def xloader_hook(context, data_dict):
         p.toolkit.get_action('xloader_submit')(
             context, {'resource_id': res_id})
 
+    utils.send_xloader_status(task)
+
 
 @side_effect_free
 def xloader_status(context, data_dict):

ckanext/xloader/interfaces.py

@@ -47,3 +47,20 @@ def after_upload(self, context, resource_dict, dataset_dict):
             the resource that was uploaded
         """
         pass
+
+
+class IPipeXloader(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 xloadered, it broadcasts its status in the DataStore,
+    perhaps triggering a process which notifies users of the status.
+    These processes can in turn put the resulting resource DS status into the pipeline
+    """
+
+    def receive_xloader_status(self, xloader_status):
+        pass

ckanext/xloader/utils.py

@@ -10,6 +10,12 @@
 
 import ckan.plugins as p
 from ckan.plugins.toolkit import config
+from ckanext.xloader.interfaces import IPipeXloader
+
+from logging import getLogger
+
+
+log = getLogger(__name__)
 
 # resource.formats accepted by ckanext-xloader. Must be lowercase here.
 DEFAULT_FORMATS = [
@@ -257,3 +263,14 @@ def datastore_resource_exists(resource_id):
     except p.toolkit.ObjectNotFound:
         return False
     return response or {'fields': []}
+
+
+def send_xloader_status(xloader_status):
+    for observer in p.PluginImplementations(IPipeXloader):
+        try:
+            observer.receive_xloader_status(xloader_status)
+        except Exception as ex:
+            log.exception(ex)
+            # We reraise all exceptions so they are obvious there
+            # is something wrong
+            raise