[#2334] Avoid duplicate Datapusher jobs for a resource. Thanks @cleme…

…ntmouchet.

ckanext/datapusher/logic/action.py

@@ -84,18 +84,33 @@ def datapusher_submit(context, data_dict):
         'error': '{}',
     }
     try:
-        task_id = p.toolkit.get_action('task_status_show')(context, {
+        existing_task = p.toolkit.get_action('task_status_show')(context, {
             'entity_id': res_id,
             'task_type': 'datapusher',
             'key': 'datapusher'
-        })['id']
-        task['id'] = task_id
+        })
+
+        if existing_task and existing_task.get('state') == 'pending':
+            updated = datetime.datetime.strptime(
+                existing_task['last_updated'], '%Y-%m-%dT%H:%M:%S.%f')
+            time_since_last_updated = datetime.datetime.now() - updated
+            if time_since_last_updated > datetime.timedelta(days=1):
+                # it's been a while since the job was last updated - it's more
+                # likely something went wrong with it and the state wasn't
+                # updated than its still in progress. Let it be restarted.
+                log.info('A pending task was found %r, but it is only %s hours'
+                         'old', existing_task['id'], time_since_last_updated)
+            else:
+                log.info('A pending task was found %s for this resource, so '
+                         'skipping this duplicate task', existing_task['id'])
+                return False
+
+        task['id'] = existing_task['id']
     except logic.NotFound:
         pass
 
     context['ignore_auth'] = True
-    result = p.toolkit.get_action('task_status_update')(context, task)
-    task_id = result['id']
+    p.toolkit.get_action('task_status_update')(context, task)
 
     try:
         r = requests.post(