We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Process completed but unable to post to result_url
### CKAN Version if known (or site URL)
version 2.8.1

### Please describe the expected behaviour
I install ckan with docker, and everything works fine but when I add a resource in datastore it does not work I get this error Process completed but unable to post to result_url

### Please describe the actual behaviour


### What steps can be taken to reproduce the issue? 


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ckanext/datapusher/logic/action.py`
Content:
```
1 # encoding: utf-8
2 
3 import logging
4 import json
5 import urlparse
6 import datetime
7 import time
8 
9 from dateutil.parser import parse as parse_date
10 
11 import requests
12 
13 import ckan.lib.helpers as h
14 import ckan.lib.navl.dictization_functions
15 import ckan.logic as logic
16 import ckan.plugins as p
17 from ckan.common import config
18 import ckanext.datapusher.logic.schema as dpschema
19 import ckanext.datapusher.interfaces as interfaces
20 
21 log = logging.getLogger(__name__)
22 _get_or_bust = logic.get_or_bust
23 _validate = ckan.lib.navl.dictization_functions.validate
24 
25 
26 def datapusher_submit(context, data_dict):
27     ''' Submit a job to the datapusher. The datapusher is a service that
28     imports tabular data into the datastore.
29 
30     :param resource_id: The resource id of the resource that the data
31         should be imported in. The resource's URL will be used to get the data.
32     :type resource_id: string
33     :param set_url_type: If set to True, the ``url_type`` of the resource will
34         be set to ``datastore`` and the resource URL will automatically point
35         to the :ref:`datastore dump <dump>` URL. (optional, default: False)
36     :type set_url_type: bool
37     :param ignore_hash: If set to True, the datapusher will reload the file
38         even if it haven't changed. (optional, default: False)
39     :type ignore_hash: bool
40 
41     Returns ``True`` if the job has been submitted and ``False`` if the job
42     has not been submitted, i.e. when the datapusher is not configured.
43 
44     :rtype: bool
45     '''
46     schema = context.get('schema', dpschema.datapusher_submit_schema())
47     data_dict, errors = _validate(data_dict, schema, context)
48     if errors:
49         raise p.toolkit.ValidationError(errors)
50 
51     res_id = data_dict['resource_id']
52 
53     p.toolkit.check_access('datapusher_submit', context, data_dict)
54 
55     try:
56         resource_dict = p.toolkit.get_action('resource_show')(context, {
57             'id': res_id,
58         })
59     except logic.NotFound:
60         return False
61 
62     datapusher_url = config.get('ckan.datapusher.url')
63 
64     site_url = h.url_for('/', qualified=True)
65     callback_url = h.url_for('/api/3/action/datapusher_hook', qualified=True)
66 
67     user = p.toolkit.get_action('user_show')(context, {'id': context['user']})
68 
69     for plugin in p.PluginImplementations(interfaces.IDataPusher):
70         upload = plugin.can_upload(res_id)
71         if not upload:
72             msg = "Plugin {0} rejected resource {1}"\
73                 .format(plugin.__class__.__name__, res_id)
74             log.info(msg)
75             return False
76 
77     task = {
78         'entity_id': res_id,
79         'entity_type': 'resource',
80         'task_type': 'datapusher',
81         'last_updated': str(datetime.datetime.utcnow()),
82         'state': 'submitting',
83         'key': 'datapusher',
84         'value': '{}',
85         'error': '{}',
86     }
87     try:
88         existing_task = p.toolkit.get_action('task_status_show')(context, {
89             'entity_id': res_id,
90             'task_type': 'datapusher',
91             'key': 'datapusher'
92         })
93         assume_task_stale_after = datetime.timedelta(seconds=int(
94             config.get('ckan.datapusher.assume_task_stale_after', 3600)))
95         if existing_task.get('state') == 'pending':
96             updated = datetime.datetime.strptime(
97                 existing_task['last_updated'], '%Y-%m-%dT%H:%M:%S.%f')
98             time_since_last_updated = datetime.datetime.utcnow() - updated
99             if time_since_last_updated > assume_task_stale_after:
100                 # it's been a while since the job was last updated - it's more
101                 # likely something went wrong with it and the state wasn't
102                 # updated than its still in progress. Let it be restarted.
103                 log.info('A pending task was found %r, but it is only %s hours'
104                          ' old', existing_task['id'], time_since_last_updated)
105             else:
106                 log.info('A pending task was found %s for this resource, so '
107                          'skipping this duplicate task', existing_task['id'])
108                 return False
109 
110         task['id'] = existing_task['id']
111     except logic.NotFound:
112         pass
113 
114     context['ignore_auth'] = True
115     p.toolkit.get_action('task_status_update')(context, task)
116 
117     try:
118         r = requests.post(
119             urlparse.urljoin(datapusher_url, 'job'),
120             headers={
121                 'Content-Type': 'application/json'
122             },
123             data=json.dumps({
124                 'api_key': user['apikey'],
125                 'job_type': 'push_to_datastore',
126                 'result_url': callback_url,
127                 'metadata': {
128                     'ignore_hash': data_dict.get('ignore_hash', False),
129                     'ckan_url': site_url,
130                     'resource_id': res_id,
131                     'set_url_type': data_dict.get('set_url_type', False),
132                     'task_created': task['last_updated'],
133                     'original_url': resource_dict.get('url'),
134                 }
135             }))
136         r.raise_for_status()
137     except requests.exceptions.ConnectionError as e:
138         error = {'message': 'Could not connect to DataPusher.',
139                  'details': str(e)}
140         task['error'] = json.dumps(error)
141         task['state'] = 'error'
142         task['last_updated'] = str(datetime.datetime.utcnow()),
143         p.toolkit.get_action('task_status_update')(context, task)
144         raise p.toolkit.ValidationError(error)
145 
146     except requests.exceptions.HTTPError as e:
147         m = 'An Error occurred while sending the job: {0}'.format(e.message)
148         try:
149             body = e.response.json()
150         except ValueError:
151             body = e.response.text
152         error = {'message': m,
153                  'details': body,
154                  'status_code': r.status_code}
155         task['error'] = json.dumps(error)
156         task['state'] = 'error'
157         task['last_updated'] = str(datetime.datetime.utcnow()),
158         p.toolkit.get_action('task_status_update')(context, task)
159         raise p.toolkit.ValidationError(error)
160 
161     value = json.dumps({'job_id': r.json()['job_id'],
162                         'job_key': r.json()['job_key']})
163 
164     task['value'] = value
165     task['state'] = 'pending'
166     task['last_updated'] = str(datetime.datetime.utcnow()),
167     p.toolkit.get_action('task_status_update')(context, task)
168 
169     return True
170 
171 
172 def datapusher_hook(context, data_dict):
173     ''' Update datapusher task. This action is typically called by the
174     datapusher whenever the status of a job changes.
175 
176     :param metadata: metadata produced by datapuser service must have
177        resource_id property.
178     :type metadata: dict
179     :param status: status of the job from the datapusher service
180     :type status: string
181     '''
182 
183     metadata, status = _get_or_bust(data_dict, ['metadata', 'status'])
184 
185     res_id = _get_or_bust(metadata, 'resource_id')
186 
187     # Pass metadata, not data_dict, as it contains the resource id needed
188     # on the auth checks
189     p.toolkit.check_access('datapusher_submit', context, metadata)
190 
191     task = p.toolkit.get_action('task_status_show')(context, {
192         'entity_id': res_id,
193         'task_type': 'datapusher',
194         'key': 'datapusher'
195     })
196 
197     task['state'] = status
198     task['last_updated'] = str(datetime.datetime.utcnow())
199 
200     resubmit = False
201 
202     if status == 'complete':
203         # Create default views for resource if necessary (only the ones that
204         # require data to be in the DataStore)
205         resource_dict = p.toolkit.get_action('resource_show')(
206             context, {'id': res_id})
207 
208         dataset_dict = p.toolkit.get_action('package_show')(
209             context, {'id': resource_dict['package_id']})
210 
211         for plugin in p.PluginImplementations(interfaces.IDataPusher):
212             plugin.after_upload(context, resource_dict, dataset_dict)
213 
214         logic.get_action('resource_create_default_resource_views')(
215             context,
216             {
217                 'resource': resource_dict,
218                 'package': dataset_dict,
219                 'create_datastore_views': True,
220             })
221 
222         # Check if the uploaded file has been modified in the meantime
223         if (resource_dict.get('last_modified') and
224                 metadata.get('task_created')):
225             try:
226                 last_modified_datetime = parse_date(
227                     resource_dict['last_modified'])
228                 task_created_datetime = parse_date(metadata['task_created'])
229                 if last_modified_datetime > task_created_datetime:
230                     log.debug('Uploaded file more recent: {0} > {1}'.format(
231                         last_modified_datetime, task_created_datetime))
232                     resubmit = True
233             except ValueError:
234                 pass
235         # Check if the URL of the file has been modified in the meantime
236         elif (resource_dict.get('url') and
237                 metadata.get('original_url') and
238                 resource_dict['url'] != metadata['original_url']):
239             log.debug('URLs are different: {0} != {1}'.format(
240                 resource_dict['url'], metadata['original_url']))
241             resubmit = True
242 
243     context['ignore_auth'] = True
244     p.toolkit.get_action('task_status_update')(context, task)
245 
246     if resubmit:
247         log.debug('Resource {0} has been modified, '
248                   'resubmitting to DataPusher'.format(res_id))
249         p.toolkit.get_action('datapusher_submit')(
250             context, {'resource_id': res_id})
251 
252 
253 def datapusher_status(context, data_dict):
254     ''' Get the status of a datapusher job for a certain resource.
255 
256     :param resource_id: The resource id of the resource that you want the
257         datapusher status for.
258     :type resource_id: string
259     '''
260 
261     p.toolkit.check_access('datapusher_status', context, data_dict)
262 
263     if 'id' in data_dict:
264         data_dict['resource_id'] = data_dict['id']
265     res_id = _get_or_bust(data_dict, 'resource_id')
266 
267     task = p.toolkit.get_action('task_status_show')(context, {
268         'entity_id': res_id,
269         'task_type': 'datapusher',
270         'key': 'datapusher'
271     })
272 
273     datapusher_url = config.get('ckan.datapusher.url')
274     if not datapusher_url:
275         raise p.toolkit.ValidationError(
276             {'configuration': ['ckan.datapusher.url not in config file']})
277 
278     value = json.loads(task['value'])
279     job_key = value.get('job_key')
280     job_id = value.get('job_id')
281     url = None
282     job_detail = None
283 
284     if job_id:
285         url = urlparse.urljoin(datapusher_url, 'job' + '/' + job_id)
286         try:
287             r = requests.get(url, headers={'Content-Type': 'application/json',
288                                            'Authorization': job_key})
289             r.raise_for_status()
290             job_detail = r.json()
291             for log in job_detail['logs']:
292                 if 'timestamp' in log:
293                     date = time.strptime(
294                         log['timestamp'], "%Y-%m-%dT%H:%M:%S.%f")
295                     date = datetime.datetime.utcfromtimestamp(
296                         time.mktime(date))
297                     log['timestamp'] = date
298         except (requests.exceptions.ConnectionError,
299                 requests.exceptions.HTTPError):
300             job_detail = {'error': 'cannot connect to datapusher'}
301 
302     return {
303         'status': task['state'],
304         'job_id': job_id,
305         'job_url': url,
306         'last_updated': task['last_updated'],
307         'job_key': job_key,
308         'task_info': job_detail,
309         'error': json.loads(task['error'])
310     }
311
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/ckanext/datapusher/logic/action.py b/ckanext/datapusher/logic/action.py
--- a/ckanext/datapusher/logic/action.py
+++ b/ckanext/datapusher/logic/action.py
@@ -62,7 +62,14 @@
     datapusher_url = config.get('ckan.datapusher.url')
 
     site_url = h.url_for('/', qualified=True)
-    callback_url = h.url_for('/api/3/action/datapusher_hook', qualified=True)
+
+    callback_url_base = config.get('ckan.datapusher.callback_url_base')
+    if callback_url_base:
+        callback_url = urlparse.urljoin(
+            callback_url_base.rstrip('/'), '/api/3/action/datapusher_hook')
+    else:
+        callback_url = h.url_for(
+            '/api/3/action/datapusher_hook', qualified=True)
 
     user = p.toolkit.get_action('user_show')(context, {'id': context['user']})

{"golden_diff": "diff --git a/ckanext/datapusher/logic/action.py b/ckanext/datapusher/logic/action.py\n--- a/ckanext/datapusher/logic/action.py\n+++ b/ckanext/datapusher/logic/action.py\n@@ -62,7 +62,14 @@\n     datapusher_url = config.get('ckan.datapusher.url')\n \n     site_url = h.url_for('/', qualified=True)\n-    callback_url = h.url_for('/api/3/action/datapusher_hook', qualified=True)\n+\n+    callback_url_base = config.get('ckan.datapusher.callback_url_base')\n+    if callback_url_base:\n+        callback_url = urlparse.urljoin(\n+            callback_url_base.rstrip('/'), '/api/3/action/datapusher_hook')\n+    else:\n+        callback_url = h.url_for(\n+            '/api/3/action/datapusher_hook', qualified=True)\n \n     user = p.toolkit.get_action('user_show')(context, {'id': context['user']})\n", "issue": "Process completed but unable to post to result_url\n### CKAN Version if known (or site URL)\r\nversion 2.8.1\r\n\r\n### Please describe the expected behaviour\r\nI install ckan with docker, and everything works fine but when I add a resource in datastore it does not work I get this error Process completed but unable to post to result_url\r\n\r\n### Please describe the actual behaviour\r\n\r\n\r\n### What steps can be taken to reproduce the issue? \r\n\n", "before_files": [{"content": "# encoding: utf-8\n\nimport logging\nimport json\nimport urlparse\nimport datetime\nimport time\n\nfrom dateutil.parser import parse as parse_date\n\nimport requests\n\nimport ckan.lib.helpers as h\nimport ckan.lib.navl.dictization_functions\nimport ckan.logic as logic\nimport ckan.plugins as p\nfrom ckan.common import config\nimport ckanext.datapusher.logic.schema as dpschema\nimport ckanext.datapusher.interfaces as interfaces\n\nlog = logging.getLogger(__name__)\n_get_or_bust = logic.get_or_bust\n_validate = ckan.lib.navl.dictization_functions.validate\n\n\ndef datapusher_submit(context, data_dict):\n    ''' Submit a job to the datapusher. The datapusher is a service that\n    imports tabular data into the datastore.\n\n    :param resource_id: The resource id of the resource that the data\n        should be imported in. The resource's URL will be used to get the data.\n    :type resource_id: string\n    :param set_url_type: If set to True, the ``url_type`` of the resource will\n        be set to ``datastore`` and the resource URL will automatically point\n        to the :ref:`datastore dump <dump>` URL. (optional, default: False)\n    :type set_url_type: bool\n    :param ignore_hash: If set to True, the datapusher will reload the file\n        even if it haven't changed. (optional, default: False)\n    :type ignore_hash: bool\n\n    Returns ``True`` if the job has been submitted and ``False`` if the job\n    has not been submitted, i.e. when the datapusher is not configured.\n\n    :rtype: bool\n    '''\n    schema = context.get('schema', dpschema.datapusher_submit_schema())\n    data_dict, errors = _validate(data_dict, schema, context)\n    if errors:\n        raise p.toolkit.ValidationError(errors)\n\n    res_id = data_dict['resource_id']\n\n    p.toolkit.check_access('datapusher_submit', context, data_dict)\n\n    try:\n        resource_dict = p.toolkit.get_action('resource_show')(context, {\n            'id': res_id,\n        })\n    except logic.NotFound:\n        return False\n\n    datapusher_url = config.get('ckan.datapusher.url')\n\n    site_url = h.url_for('/', qualified=True)\n    callback_url = h.url_for('/api/3/action/datapusher_hook', qualified=True)\n\n    user = p.toolkit.get_action('user_show')(context, {'id': context['user']})\n\n    for plugin in p.PluginImplementations(interfaces.IDataPusher):\n        upload = plugin.can_upload(res_id)\n        if not upload:\n            msg = \"Plugin {0} rejected resource {1}\"\\\n                .format(plugin.__class__.__name__, res_id)\n            log.info(msg)\n            return False\n\n    task = {\n        'entity_id': res_id,\n        'entity_type': 'resource',\n        'task_type': 'datapusher',\n        'last_updated': str(datetime.datetime.utcnow()),\n        'state': 'submitting',\n        'key': 'datapusher',\n        'value': '{}',\n        'error': '{}',\n    }\n    try:\n        existing_task = p.toolkit.get_action('task_status_show')(context, {\n            'entity_id': res_id,\n            'task_type': 'datapusher',\n            'key': 'datapusher'\n        })\n        assume_task_stale_after = datetime.timedelta(seconds=int(\n            config.get('ckan.datapusher.assume_task_stale_after', 3600)))\n        if existing_task.get('state') == 'pending':\n            updated = datetime.datetime.strptime(\n                existing_task['last_updated'], '%Y-%m-%dT%H:%M:%S.%f')\n            time_since_last_updated = datetime.datetime.utcnow() - updated\n            if time_since_last_updated > assume_task_stale_after:\n                # it's been a while since the job was last updated - it's more\n                # likely something went wrong with it and the state wasn't\n                # updated than its still in progress. Let it be restarted.\n                log.info('A pending task was found %r, but it is only %s hours'\n                         ' old', existing_task['id'], time_since_last_updated)\n            else:\n                log.info('A pending task was found %s for this resource, so '\n                         'skipping this duplicate task', existing_task['id'])\n                return False\n\n        task['id'] = existing_task['id']\n    except logic.NotFound:\n        pass\n\n    context['ignore_auth'] = True\n    p.toolkit.get_action('task_status_update')(context, task)\n\n    try:\n        r = requests.post(\n            urlparse.urljoin(datapusher_url, 'job'),\n            headers={\n                'Content-Type': 'application/json'\n            },\n            data=json.dumps({\n                'api_key': user['apikey'],\n                'job_type': 'push_to_datastore',\n                'result_url': callback_url,\n                'metadata': {\n                    'ignore_hash': data_dict.get('ignore_hash', False),\n                    'ckan_url': site_url,\n                    'resource_id': res_id,\n                    'set_url_type': data_dict.get('set_url_type', False),\n                    'task_created': task['last_updated'],\n                    'original_url': resource_dict.get('url'),\n                }\n            }))\n        r.raise_for_status()\n    except requests.exceptions.ConnectionError as e:\n        error = {'message': 'Could not connect to DataPusher.',\n                 'details': str(e)}\n        task['error'] = json.dumps(error)\n        task['state'] = 'error'\n        task['last_updated'] = str(datetime.datetime.utcnow()),\n        p.toolkit.get_action('task_status_update')(context, task)\n        raise p.toolkit.ValidationError(error)\n\n    except requests.exceptions.HTTPError as e:\n        m = 'An Error occurred while sending the job: {0}'.format(e.message)\n        try:\n            body = e.response.json()\n        except ValueError:\n            body = e.response.text\n        error = {'message': m,\n                 'details': body,\n                 'status_code': r.status_code}\n        task['error'] = json.dumps(error)\n        task['state'] = 'error'\n        task['last_updated'] = str(datetime.datetime.utcnow()),\n        p.toolkit.get_action('task_status_update')(context, task)\n        raise p.toolkit.ValidationError(error)\n\n    value = json.dumps({'job_id': r.json()['job_id'],\n                        'job_key': r.json()['job_key']})\n\n    task['value'] = value\n    task['state'] = 'pending'\n    task['last_updated'] = str(datetime.datetime.utcnow()),\n    p.toolkit.get_action('task_status_update')(context, task)\n\n    return True\n\n\ndef datapusher_hook(context, data_dict):\n    ''' Update datapusher task. This action is typically called by the\n    datapusher whenever the status of a job changes.\n\n    :param metadata: metadata produced by datapuser service must have\n       resource_id property.\n    :type metadata: dict\n    :param status: status of the job from the datapusher service\n    :type status: string\n    '''\n\n    metadata, status = _get_or_bust(data_dict, ['metadata', 'status'])\n\n    res_id = _get_or_bust(metadata, 'resource_id')\n\n    # Pass metadata, not data_dict, as it contains the resource id needed\n    # on the auth checks\n    p.toolkit.check_access('datapusher_submit', context, metadata)\n\n    task = p.toolkit.get_action('task_status_show')(context, {\n        'entity_id': res_id,\n        'task_type': 'datapusher',\n        'key': 'datapusher'\n    })\n\n    task['state'] = status\n    task['last_updated'] = str(datetime.datetime.utcnow())\n\n    resubmit = False\n\n    if status == 'complete':\n        # Create default views for resource if necessary (only the ones that\n        # require data to be in the DataStore)\n        resource_dict = p.toolkit.get_action('resource_show')(\n            context, {'id': res_id})\n\n        dataset_dict = p.toolkit.get_action('package_show')(\n            context, {'id': resource_dict['package_id']})\n\n        for plugin in p.PluginImplementations(interfaces.IDataPusher):\n            plugin.after_upload(context, resource_dict, dataset_dict)\n\n        logic.get_action('resource_create_default_resource_views')(\n            context,\n            {\n                'resource': resource_dict,\n                'package': dataset_dict,\n                'create_datastore_views': True,\n            })\n\n        # Check if the uploaded file has been modified in the meantime\n        if (resource_dict.get('last_modified') and\n                metadata.get('task_created')):\n            try:\n                last_modified_datetime = parse_date(\n                    resource_dict['last_modified'])\n                task_created_datetime = parse_date(metadata['task_created'])\n                if last_modified_datetime > task_created_datetime:\n                    log.debug('Uploaded file more recent: {0} > {1}'.format(\n                        last_modified_datetime, task_created_datetime))\n                    resubmit = True\n            except ValueError:\n                pass\n        # Check if the URL of the file has been modified in the meantime\n        elif (resource_dict.get('url') and\n                metadata.get('original_url') and\n                resource_dict['url'] != metadata['original_url']):\n            log.debug('URLs are different: {0} != {1}'.format(\n                resource_dict['url'], metadata['original_url']))\n            resubmit = True\n\n    context['ignore_auth'] = True\n    p.toolkit.get_action('task_status_update')(context, task)\n\n    if resubmit:\n        log.debug('Resource {0} has been modified, '\n                  'resubmitting to DataPusher'.format(res_id))\n        p.toolkit.get_action('datapusher_submit')(\n            context, {'resource_id': res_id})\n\n\ndef datapusher_status(context, data_dict):\n    ''' Get the status of a datapusher job for a certain resource.\n\n    :param resource_id: The resource id of the resource that you want the\n        datapusher status for.\n    :type resource_id: string\n    '''\n\n    p.toolkit.check_access('datapusher_status', context, data_dict)\n\n    if 'id' in data_dict:\n        data_dict['resource_id'] = data_dict['id']\n    res_id = _get_or_bust(data_dict, 'resource_id')\n\n    task = p.toolkit.get_action('task_status_show')(context, {\n        'entity_id': res_id,\n        'task_type': 'datapusher',\n        'key': 'datapusher'\n    })\n\n    datapusher_url = config.get('ckan.datapusher.url')\n    if not datapusher_url:\n        raise p.toolkit.ValidationError(\n            {'configuration': ['ckan.datapusher.url not in config file']})\n\n    value = json.loads(task['value'])\n    job_key = value.get('job_key')\n    job_id = value.get('job_id')\n    url = None\n    job_detail = None\n\n    if job_id:\n        url = urlparse.urljoin(datapusher_url, 'job' + '/' + job_id)\n        try:\n            r = requests.get(url, headers={'Content-Type': 'application/json',\n                                           'Authorization': job_key})\n            r.raise_for_status()\n            job_detail = r.json()\n            for log in job_detail['logs']:\n                if 'timestamp' in log:\n                    date = time.strptime(\n                        log['timestamp'], \"%Y-%m-%dT%H:%M:%S.%f\")\n                    date = datetime.datetime.utcfromtimestamp(\n                        time.mktime(date))\n                    log['timestamp'] = date\n        except (requests.exceptions.ConnectionError,\n                requests.exceptions.HTTPError):\n            job_detail = {'error': 'cannot connect to datapusher'}\n\n    return {\n        'status': task['state'],\n        'job_id': job_id,\n        'job_url': url,\n        'last_updated': task['last_updated'],\n        'job_key': job_key,\n        'task_info': job_detail,\n        'error': json.loads(task['error'])\n    }\n", "path": "ckanext/datapusher/logic/action.py"}], "after_files": [{"content": "# encoding: utf-8\n\nimport logging\nimport json\nimport urlparse\nimport datetime\nimport time\n\nfrom dateutil.parser import parse as parse_date\n\nimport requests\n\nimport ckan.lib.helpers as h\nimport ckan.lib.navl.dictization_functions\nimport ckan.logic as logic\nimport ckan.plugins as p\nfrom ckan.common import config\nimport ckanext.datapusher.logic.schema as dpschema\nimport ckanext.datapusher.interfaces as interfaces\n\nlog = logging.getLogger(__name__)\n_get_or_bust = logic.get_or_bust\n_validate = ckan.lib.navl.dictization_functions.validate\n\n\ndef datapusher_submit(context, data_dict):\n    ''' Submit a job to the datapusher. The datapusher is a service that\n    imports tabular data into the datastore.\n\n    :param resource_id: The resource id of the resource that the data\n        should be imported in. The resource's URL will be used to get the data.\n    :type resource_id: string\n    :param set_url_type: If set to True, the ``url_type`` of the resource will\n        be set to ``datastore`` and the resource URL will automatically point\n        to the :ref:`datastore dump <dump>` URL. (optional, default: False)\n    :type set_url_type: bool\n    :param ignore_hash: If set to True, the datapusher will reload the file\n        even if it haven't changed. (optional, default: False)\n    :type ignore_hash: bool\n\n    Returns ``True`` if the job has been submitted and ``False`` if the job\n    has not been submitted, i.e. when the datapusher is not configured.\n\n    :rtype: bool\n    '''\n    schema = context.get('schema', dpschema.datapusher_submit_schema())\n    data_dict, errors = _validate(data_dict, schema, context)\n    if errors:\n        raise p.toolkit.ValidationError(errors)\n\n    res_id = data_dict['resource_id']\n\n    p.toolkit.check_access('datapusher_submit', context, data_dict)\n\n    try:\n        resource_dict = p.toolkit.get_action('resource_show')(context, {\n            'id': res_id,\n        })\n    except logic.NotFound:\n        return False\n\n    datapusher_url = config.get('ckan.datapusher.url')\n\n    site_url = h.url_for('/', qualified=True)\n\n    callback_url_base = config.get('ckan.datapusher.callback_url_base')\n    if callback_url_base:\n        callback_url = urlparse.urljoin(\n            callback_url_base.rstrip('/'), '/api/3/action/datapusher_hook')\n    else:\n        callback_url = h.url_for(\n            '/api/3/action/datapusher_hook', qualified=True)\n\n    user = p.toolkit.get_action('user_show')(context, {'id': context['user']})\n\n    for plugin in p.PluginImplementations(interfaces.IDataPusher):\n        upload = plugin.can_upload(res_id)\n        if not upload:\n            msg = \"Plugin {0} rejected resource {1}\"\\\n                .format(plugin.__class__.__name__, res_id)\n            log.info(msg)\n            return False\n\n    task = {\n        'entity_id': res_id,\n        'entity_type': 'resource',\n        'task_type': 'datapusher',\n        'last_updated': str(datetime.datetime.utcnow()),\n        'state': 'submitting',\n        'key': 'datapusher',\n        'value': '{}',\n        'error': '{}',\n    }\n    try:\n        existing_task = p.toolkit.get_action('task_status_show')(context, {\n            'entity_id': res_id,\n            'task_type': 'datapusher',\n            'key': 'datapusher'\n        })\n        assume_task_stale_after = datetime.timedelta(seconds=int(\n            config.get('ckan.datapusher.assume_task_stale_after', 3600)))\n        if existing_task.get('state') == 'pending':\n            updated = datetime.datetime.strptime(\n                existing_task['last_updated'], '%Y-%m-%dT%H:%M:%S.%f')\n            time_since_last_updated = datetime.datetime.utcnow() - updated\n            if time_since_last_updated > assume_task_stale_after:\n                # it's been a while since the job was last updated - it's more\n                # likely something went wrong with it and the state wasn't\n                # updated than its still in progress. Let it be restarted.\n                log.info('A pending task was found %r, but it is only %s hours'\n                         ' old', existing_task['id'], time_since_last_updated)\n            else:\n                log.info('A pending task was found %s for this resource, so '\n                         'skipping this duplicate task', existing_task['id'])\n                return False\n\n        task['id'] = existing_task['id']\n    except logic.NotFound:\n        pass\n\n    context['ignore_auth'] = True\n    p.toolkit.get_action('task_status_update')(context, task)\n\n    try:\n        r = requests.post(\n            urlparse.urljoin(datapusher_url, 'job'),\n            headers={\n                'Content-Type': 'application/json'\n            },\n            data=json.dumps({\n                'api_key': user['apikey'],\n                'job_type': 'push_to_datastore',\n                'result_url': callback_url,\n                'metadata': {\n                    'ignore_hash': data_dict.get('ignore_hash', False),\n                    'ckan_url': site_url,\n                    'resource_id': res_id,\n                    'set_url_type': data_dict.get('set_url_type', False),\n                    'task_created': task['last_updated'],\n                    'original_url': resource_dict.get('url'),\n                }\n            }))\n        r.raise_for_status()\n    except requests.exceptions.ConnectionError as e:\n        error = {'message': 'Could not connect to DataPusher.',\n                 'details': str(e)}\n        task['error'] = json.dumps(error)\n        task['state'] = 'error'\n        task['last_updated'] = str(datetime.datetime.utcnow()),\n        p.toolkit.get_action('task_status_update')(context, task)\n        raise p.toolkit.ValidationError(error)\n\n    except requests.exceptions.HTTPError as e:\n        m = 'An Error occurred while sending the job: {0}'.format(e.message)\n        try:\n            body = e.response.json()\n        except ValueError:\n            body = e.response.text\n        error = {'message': m,\n                 'details': body,\n                 'status_code': r.status_code}\n        task['error'] = json.dumps(error)\n        task['state'] = 'error'\n        task['last_updated'] = str(datetime.datetime.utcnow()),\n        p.toolkit.get_action('task_status_update')(context, task)\n        raise p.toolkit.ValidationError(error)\n\n    value = json.dumps({'job_id': r.json()['job_id'],\n                        'job_key': r.json()['job_key']})\n\n    task['value'] = value\n    task['state'] = 'pending'\n    task['last_updated'] = str(datetime.datetime.utcnow()),\n    p.toolkit.get_action('task_status_update')(context, task)\n\n    return True\n\n\ndef datapusher_hook(context, data_dict):\n    ''' Update datapusher task. This action is typically called by the\n    datapusher whenever the status of a job changes.\n\n    :param metadata: metadata produced by datapuser service must have\n       resource_id property.\n    :type metadata: dict\n    :param status: status of the job from the datapusher service\n    :type status: string\n    '''\n\n    metadata, status = _get_or_bust(data_dict, ['metadata', 'status'])\n\n    res_id = _get_or_bust(metadata, 'resource_id')\n\n    # Pass metadata, not data_dict, as it contains the resource id needed\n    # on the auth checks\n    p.toolkit.check_access('datapusher_submit', context, metadata)\n\n    task = p.toolkit.get_action('task_status_show')(context, {\n        'entity_id': res_id,\n        'task_type': 'datapusher',\n        'key': 'datapusher'\n    })\n\n    task['state'] = status\n    task['last_updated'] = str(datetime.datetime.utcnow())\n\n    resubmit = False\n\n    if status == 'complete':\n        # Create default views for resource if necessary (only the ones that\n        # require data to be in the DataStore)\n        resource_dict = p.toolkit.get_action('resource_show')(\n            context, {'id': res_id})\n\n        dataset_dict = p.toolkit.get_action('package_show')(\n            context, {'id': resource_dict['package_id']})\n\n        for plugin in p.PluginImplementations(interfaces.IDataPusher):\n            plugin.after_upload(context, resource_dict, dataset_dict)\n\n        logic.get_action('resource_create_default_resource_views')(\n            context,\n            {\n                'resource': resource_dict,\n                'package': dataset_dict,\n                'create_datastore_views': True,\n            })\n\n        # Check if the uploaded file has been modified in the meantime\n        if (resource_dict.get('last_modified') and\n                metadata.get('task_created')):\n            try:\n                last_modified_datetime = parse_date(\n                    resource_dict['last_modified'])\n                task_created_datetime = parse_date(metadata['task_created'])\n                if last_modified_datetime > task_created_datetime:\n                    log.debug('Uploaded file more recent: {0} > {1}'.format(\n                        last_modified_datetime, task_created_datetime))\n                    resubmit = True\n            except ValueError:\n                pass\n        # Check if the URL of the file has been modified in the meantime\n        elif (resource_dict.get('url') and\n                metadata.get('original_url') and\n                resource_dict['url'] != metadata['original_url']):\n            log.debug('URLs are different: {0} != {1}'.format(\n                resource_dict['url'], metadata['original_url']))\n            resubmit = True\n\n    context['ignore_auth'] = True\n    p.toolkit.get_action('task_status_update')(context, task)\n\n    if resubmit:\n        log.debug('Resource {0} has been modified, '\n                  'resubmitting to DataPusher'.format(res_id))\n        p.toolkit.get_action('datapusher_submit')(\n            context, {'resource_id': res_id})\n\n\ndef datapusher_status(context, data_dict):\n    ''' Get the status of a datapusher job for a certain resource.\n\n    :param resource_id: The resource id of the resource that you want the\n        datapusher status for.\n    :type resource_id: string\n    '''\n\n    p.toolkit.check_access('datapusher_status', context, data_dict)\n\n    if 'id' in data_dict:\n        data_dict['resource_id'] = data_dict['id']\n    res_id = _get_or_bust(data_dict, 'resource_id')\n\n    task = p.toolkit.get_action('task_status_show')(context, {\n        'entity_id': res_id,\n        'task_type': 'datapusher',\n        'key': 'datapusher'\n    })\n\n    datapusher_url = config.get('ckan.datapusher.url')\n    if not datapusher_url:\n        raise p.toolkit.ValidationError(\n            {'configuration': ['ckan.datapusher.url not in config file']})\n\n    value = json.loads(task['value'])\n    job_key = value.get('job_key')\n    job_id = value.get('job_id')\n    url = None\n    job_detail = None\n\n    if job_id:\n        url = urlparse.urljoin(datapusher_url, 'job' + '/' + job_id)\n        try:\n            r = requests.get(url, headers={'Content-Type': 'application/json',\n                                           'Authorization': job_key})\n            r.raise_for_status()\n            job_detail = r.json()\n            for log in job_detail['logs']:\n                if 'timestamp' in log:\n                    date = time.strptime(\n                        log['timestamp'], \"%Y-%m-%dT%H:%M:%S.%f\")\n                    date = datetime.datetime.utcfromtimestamp(\n                        time.mktime(date))\n                    log['timestamp'] = date\n        except (requests.exceptions.ConnectionError,\n                requests.exceptions.HTTPError):\n            job_detail = {'error': 'cannot connect to datapusher'}\n\n    return {\n        'status': task['state'],\n        'job_id': job_id,\n        'job_url': url,\n        'last_updated': task['last_updated'],\n        'job_key': job_key,\n        'task_info': job_detail,\n        'error': json.loads(task['error'])\n    }\n", "path": "ckanext/datapusher/logic/action.py"}]}