[#940] Add datapusher callback hook

ckanext/datastore/logic/action.py

@@ -485,6 +485,31 @@ def datapusher_submit(context, data_dict):
     return True
 
 
+def datapusher_hook(context, data_dict):
+    """ Update datapusher task. This action is typically called by the
+    datapusher whenever the status of a job changes.
+
+    Expects a job with ``status``, ``metadata``.
+    """
+
+    # TODO: use a schema to validate
+
+    p.toolkit.check_access('datapusher_submit', context, data_dict)
+
+    res_id = data_dict['metadata']['resource_id']
+
+    task = p.toolkit.get_action('task_status_show')(context, {
+        'entity_id': res_id,
+        'task_type': 'datapusher',
+        'key': 'datapusher'
+    })
+
+    task['state'] = data_dict['status']
+    task['last_updated'] = str(datetime.datetime.now())
+
+    p.toolkit.get_action('task_status_update')(context, task)
+
+
 def _resource_exists(context, data_dict):
     # Returns true if the resource exists in CKAN and in the datastore
     model = _get_or_bust(context, 'model')

ckanext/datastore/plugin.py

@@ -226,6 +226,7 @@ def get_actions(self):
                    'datastore_delete': action.datastore_delete,
                    'datastore_search': action.datastore_search,
                    'datapusher_submit': action.datapusher_submit,
+                   'datapusher_hook': action.datapusher_hook,
                    'resource_show': self.resource_show_action,
                   }
         if not self.legacy_mode:

ckanext/datastore/tests/test_create.py

@@ -2,6 +2,7 @@
 import httpretty
 import nose
 import sys
+import datetime
 from nose.tools import assert_equal
 
 import pylons
@@ -619,6 +620,45 @@ def test_send_datapusher_creates_task(self):
 
         assert task['state'] == 'pending', task
 
+    def test_datapusher_hook(self):
+        package = model.Package.get('annakarenina')
+        resource = package.resources[0]
+
+        context = {
+            'user': self.sysadmin_user.name
+        }
+
+        p.toolkit.get_action('task_status_update')(context, {
+            'entity_id': resource.id,
+            'entity_type': 'resource',
+            'task_type': 'datapusher',
+            'key': 'datapusher',
+            'value': True,
+            'last_updated': str(datetime.datetime.now()),
+            'state': 'pending'
+        })
+
+        data = {
+            'status': 'success',
+            'metadata': {
+                'resource_id': resource.id
+            }
+        }
+        postparams = '%s=1' % json.dumps(data)
+        auth = {'Authorization': str(self.sysadmin_user.apikey)}
+        res = self.app.post('/api/action/datapusher_hook', params=postparams,
+                            extra_environ=auth, status=200)
+        print res.body
+        res_dict = json.loads(res.body)
+
+        assert res_dict['success'] is True
+
+        task = tests.call_action_api(
+            self.app, 'task_status_show', entity_id=resource.id,
+            task_type='datapusher', key='datapusher')
+
+        assert task['state'] == 'success', task
+
     def test_guess_types(self):
         resource = model.Package.get('annakarenina').resources[1]