Improve elastic search agent response handling

ckanext/metadata/elastic/action.py

@@ -1,7 +1,6 @@
 # encoding: utf-8
 
 import logging
-import json
 
 import ckan.plugins.toolkit as tk
 from ckan.common import _
@@ -27,10 +26,9 @@ def metadata_standard_index_create(original_action, context, data_dict):
     if metadata_standard is None:
         raise tk.ObjectNotFound('%s: %s' % (_('Not found'), _('Metadata Standard')))
 
-    log.info("Initializing search index from metadata standard %s: %r",
-             metadata_standard.name, metadata_standard.metadata_template_json)
+    log.info("Initializing search index from metadata standard %s", metadata_standard.name)
 
-    client.initialize_index(metadata_standard.name, metadata_standard.metadata_template_json)
+    return client.initialize_index(metadata_standard.name, metadata_standard.metadata_template_json)
 
 
 @tk.chained_action
@@ -50,7 +48,7 @@ def metadata_standard_index_delete(original_action, context, data_dict):
 
     log.info("Deleting search index for metadata standard %s", metadata_standard.name)
 
-    client.delete_index(metadata_standard.name)
+    return client.delete_index(metadata_standard.name)
 
 
 @tk.chained_action

ckanext/metadata/elastic/client.py

@@ -13,41 +13,58 @@
 log = logging.getLogger(__name__)
 
 
+def _search_agent_url():
+    return config.get('ckan.metadata.elastic.search_agent_url')
+
+
 @app.task
-def _call_agent(url, **kwargs):
-    response = requests.post(url, data=kwargs)
-    response.raise_for_status()
-    result = response.json()
-    # note: exceptions are lost for async calls as we're not currently using a results backend
-    # todo: for synchronous calls exception handling needs to be improved; this causes internal server error
-    if not result['success']:
-        raise Exception(result['msg'])
+def _call_agent_async(url, **kwargs):
+    _call_agent.delay(url, **kwargs)
 
 
-def _search_agent_url():
-    return config.get('ckan.metadata.elastic.search_agent_url')
+def _call_agent(url, **kwargs):
+    log.debug("POST to elastic search agent %s %r", url, kwargs)
+    try:
+        response = requests.post(url, data=kwargs)
+        response.raise_for_status()
+        result = response.json()
+        if not result['success']:
+            raise ValueError(result['msg'])
+    except requests.RequestException, e:
+        log.error("Call to elastic search agent failed: %s", e)
+        result = {
+            'success': False,
+            'msg': "Call to elastic search agent failed: %s" % e,
+        }
+    except (ValueError, KeyError), e:
+        log.error("Invalid response from elastic search agent: %s", e)
+        result = {
+            'success': False,
+            'msg': "Invalid response from elastic search agent",
+        }
+    return result
 
 
 # synchronous
 def initialize_index(index_name, metadata_template_json):
     url = _search_agent_url() + '/create_index'
-    _call_agent(url, index=index_name, metadata_json=metadata_template_json)
+    return _call_agent(url, index=index_name, metadata_json=metadata_template_json)
 
 
 # synchronous
 def delete_index(index_name):
     url = _search_agent_url() + '/delete_index'
-    _call_agent(url, index=index_name)
+    return _call_agent(url, index=index_name)
 
 
 # asynchronous
 def push_record(index_name, record_id, metadata_json, organization, collection, infrastructures):
     url = _search_agent_url() + '/add'
-    _call_agent.delay(url, index=index_name, record_id=record_id, metadata_json=metadata_json,
+    _call_agent_async(url, index=index_name, record_id=record_id, metadata_json=metadata_json,
                       organization=organization, collection=collection, infrastructures=infrastructures)
 
 
 # asynchronous
 def delete_record(index_name, record_id):
     url = _search_agent_url() + '/delete'
-    _call_agent.delay(url, index=index_name, record_id=record_id)
+    _call_agent_async(url, index=index_name, record_id=record_id)