Show datapusher status on resource page (merge #938 first)

ckan/config/deployment.ini_tmpl

@@ -133,6 +133,12 @@ ckan.feeds.author_link =
 #ofs.aws_access_key_id = ....
 #ofs.aws_secret_access_key = ....
 
+## Datapusher settings
+
+# Make sure you have set up the DataStore
+
+datapusher.formats = csv
+datapusher.url = http://datapusher.ckan.org/
 
 ## Activity Streams Settings
 

ckan/public/base/less/ckan.less

@@ -20,6 +20,7 @@
 @import "activity.less";
 @import "dropdown.less";
 @import "dashboard.less";
+@import "datapusher.less";
 
 body {
   // Using the masthead/footer gradient prevents the color from changing

ckan/public/base/less/datapusher.less

@@ -0,0 +1,18 @@
+.datapusher-status-link:hover {
+	text-decoration: none;
+}
+
+.datapusher-status {
+	&.status-unknown {
+		color: #bbb;
+	}
+	&.status-pending {
+		color: #FFCC00;
+	}
+	&.status-error {
+		color: red;
+	}
+	&.status-complete {
+		color: #009900;
+	}
+}

ckan/templates/package/resource_read.html

@@ -27,6 +27,7 @@
           <ul>
             {% block resource_actions_inner %}
             {% if h.check_access('package_update', {'id':pkg.id }) %}
+              <li>{% snippet 'snippets/datapusher_status.html', resource=res %}</li>
               <li>{% link_for _('Edit'), controller='package', action='resource_edit', id=pkg.name, resource_id=res.id, class_='btn', icon='wrench' %}</li>
             {% endif %}
             {% if res.url %}

ckan/templates/snippets/datapusher_status.html

@@ -0,0 +1,16 @@
+{# Datapusher status indicator
+
+resource: the resource
+
+#}
+{% if resource.datastore_active %}
+	{% if h.datapusher_enabled() %}
+		{% set job = h.datapusher_status(resource.id) %}
+	  {% set title =  _('Datapusher status: {status}.').format(status=job.status) %}
+	  {% if job.status == 'unknown' %}
+	  	<i title="{{ _(title) }}" class="datapusher-status status-{{ job.status }} icon-circle icon-large"></i>
+	  {% else %}
+	  	<a href="{{ job.job_url }}" class="datapusher-status-link" title="{{ _(title) }}"><i class="datapusher-status status-{{ job.status }} icon-circle icon-large"></i></a>
+  	{% endif %}
+  {% endif %}
+{% endif %}

ckanext/datastore/helpers.py

@@ -0,0 +1,15 @@
+import ckan.plugins.toolkit as toolkit
+
+
+def datapusher_status(resource_id):
+    try:
+        return toolkit.get_action('datapusher_status')(
+            {}, {'resource_id': resource_id})
+    except toolkit.ObjectNotFound:
+        return {
+            'status': 'unknown'
+        }
+
+
+def datapusher_enabled():
+    return toolkit.get_action('datapusher_enabled')({})

ckanext/datastore/logic/action.py

@@ -99,7 +99,7 @@ def datastore_create(context, data_dict):
         if has_url:
             p.toolkit.get_action('datapusher_submit')(context, {
                 'resource_id': res['id'],
-                'set_url_to_dump': True
+                'set_url_type': True
             })
         # create empty resource
         else:
@@ -384,7 +384,7 @@ def datastore_make_private(context, data_dict):
     private or a new DataStore table is created for a CKAN resource
     that belongs to a private dataset.
 
-    :param resource_id: if of resource that should become private
+    :param resource_id: id of resource that should become private
     :type resource_id: string
     '''
     if 'id' in data_dict:
@@ -429,16 +429,24 @@ def datastore_make_public(context, data_dict):
     db.make_public(context, data_dict)
 
 
+def datapusher_enabled(context, data_dict):
+    """ Returns True if the DataPusher is configured """
+
+    datapusher_url = pylons.config.get('datapusher.url')
+    return True if datapusher_url else False
+
+
 def datapusher_submit(context, data_dict):
     ''' Submit a job to the datapusher. The datapusher is a service that
     imports tabular data into the datastore.
 
     :param resource_id: The resource id of the resource that the data
         should be imported in. The resource's URL will be used to get the data.
     :type resource_id: string
-    :param set_url_to_dump: If set to true, the URL of the resource will be set
-        to the :ref:`datastore dump <dump>` URL after the data has been imported.
-    :type set_url_to_dump: boolean
+    :param set_url_type: If set to true, the ``url_type`` of the resource will
+        be set to ``datastore`` and the resource URL will automatically point
+        to the :ref:`datastore dump <dump>` URL. (optional, default: False)
+    :type set_url_type: boolean
 
     Returns ``True`` if the job has been submitted and ``False`` if the job
     has not been submitted, i.e. when the datapusher is not configured.
@@ -450,41 +458,66 @@ def datapusher_submit(context, data_dict):
         data_dict['resource_id'] = data_dict['id']
     res_id = _get_or_bust(data_dict, 'resource_id')
 
-    # TODO: handle set_url_to_dump
-
     p.toolkit.check_access('datapusher_submit', context, data_dict)
 
-    datapusher_url = pylons.config.get(
-        'datapusher.url', 'http://datapusher.ckan.org/')
-
-    # no datapusher url means the datapusher should not be used
-    if not datapusher_url:
+    if not p.toolkit.get_action('datapusher_enabled')(context):
         return False
 
+    datapusher_url = pylons.config.get('datapusher.url')
+
     callback_url = p.toolkit.url_for(
         controller='api', action='action', logic_function='datapusher_hook',
         ver=3, qualified=True)
 
     user = p.toolkit.get_action('user_show')(context, {'id': context['user']})
-    requests.post(urlparse.urljoin(datapusher_url, 'job'), data=json.dumps({
-        'api_key': user['apikey'],
-        'job_type': 'push_to_datastore',
-        'result_url': callback_url,
-        'metadata': {
-            'ckan_url': pylons.config['ckan.site_url'],
-            'resource_id': res_id
-        }
-    }))
-
-    p.toolkit.get_action('task_status_update')(context, {
+    try:
+        r = requests.post(
+            urlparse.urljoin(datapusher_url, 'job'),
+            headers={
+                'Content-Type': 'application/json'
+            },
+            data=json.dumps({
+                'api_key': user['apikey'],
+                'job_type': 'push_to_datastore',
+                'result_url': callback_url,
+                'metadata': {
+                    'ckan_url': pylons.config['ckan.site_url'],
+                    'resource_id': res_id,
+                    'set_url_type': data_dict.get('set_url_type', False)
+                }
+            }))
+        r.raise_for_status()
+    except requests.exceptions.ConnectionError, e:
+        raise p.toolkit.ValidationError({'datapusher': {
+            'message': 'Could not connect to DataPusher.',
+            'details': str(e)}})
+    except requests.exceptions.HTTPError, e:
+        m = 'An Error occurred while sending the job: {0}'.format(e.message)
+        try:
+            body = e.response.json()
+        except ValueError:
+            body = e.response.text
+        raise p.toolkit.ValidationError({'datapusher': {
+            'message': m,
+            'details': body,
+            'status_code': r.status_code}})
+
+    empty_task = {
         'entity_id': res_id,
         'entity_type': 'resource',
         'task_type': 'datapusher',
-        'key': 'datapusher',
-        'value': datapusher_url,
         'last_updated': str(datetime.datetime.now()),
         'state': 'pending'
-    })
+    }
+
+    tasks = []
+    for (k, v) in [('job_id', r.json()['job_id']),
+                   ('job_key', r.json()['job_key'])]:
+        t = empty_task.copy()
+        t['key'] = k
+        t['value'] = v
+        tasks.append(t)
+    p.toolkit.get_action('task_status_update_many')(context, {'data': tasks})
 
     return True
 
@@ -493,7 +526,7 @@ 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``.
+    Expects a job with ``status`` and ``metadata`` with a ``resource_id``.
     """
 
     # TODO: use a schema to validate
@@ -502,16 +535,66 @@ def datapusher_hook(context, data_dict):
 
     res_id = data_dict['metadata']['resource_id']
 
-    task = p.toolkit.get_action('task_status_show')(context, {
+    task_id = p.toolkit.get_action('task_status_show')(context, {
+        'entity_id': res_id,
+        'task_type': 'datapusher',
+        'key': 'job_id'
+    })
+
+    task_key = p.toolkit.get_action('task_status_show')(context, {
+        'entity_id': res_id,
+        'task_type': 'datapusher',
+        'key': 'job_key'
+    })
+
+    tasks = [task_id, task_key]
+
+    for task in tasks:
+        task['state'] = data_dict['status']
+        task['last_updated'] = str(datetime.datetime.now())
+
+    p.toolkit.get_action('task_status_update_many')(context, {'data': tasks})
+
+
+def datapusher_status(context, data_dict):
+    """ Get the status of a datapusher job for a certain resource.
+
+    :param resource_id: The resource id of the resource that you want the
+        datapusher status for.
+    :type resource_id: string
+    """
+
+    p.toolkit.check_access('datapusher_status', context, data_dict)
+
+    if 'id' in data_dict:
+        data_dict['resource_id'] = data_dict['id']
+    res_id = _get_or_bust(data_dict, 'resource_id')
+
+    task_id = p.toolkit.get_action('task_status_show')(context, {
         'entity_id': res_id,
         'task_type': 'datapusher',
-        'key': 'datapusher'
+        'key': 'job_id'
     })
 
-    task['state'] = data_dict['status']
-    task['last_updated'] = str(datetime.datetime.now())
+    task_key = p.toolkit.get_action('task_status_show')(context, {
+        'entity_id': res_id,
+        'task_type': 'datapusher',
+        'key': 'job_key'
+    })
 
-    p.toolkit.get_action('task_status_update')(context, task)
+    datapusher_url = pylons.config.get('datapusher.url')
+    if not datapusher_url:
+        raise p.toolkit.ValidationError(
+            {'configuration': ['DataPusher not configured.']})
+
+    url = urlparse.urljoin(datapusher_url, 'job' + '/' + task_id['value'])
+    return {
+        'status': task_id['state'],
+        'job_id': task_id['value'],
+        'job_url': url,
+        'last_updated': task_id['last_updated'],
+        'job_key': task_key['value']
+    }
 
 
 def _resource_exists(context, data_dict):

ckanext/datastore/logic/auth.py

@@ -38,5 +38,9 @@ def datapusher_submit(context, data_dict):
     return _datastore_auth(context, data_dict)
 
 
+def datapusher_status(context, data_dict):
+    return _datastore_auth(context, data_dict)
+
+
 def datastore_change_permissions(context, data_dict):
     return _datastore_auth(context, data_dict)