Merge branch 'postatum-94946636_reindexing_nested_rels' into develop

docs/source/making_requests.rst

@@ -4,6 +4,8 @@ Making requests
 Query syntax
 ------------
 
+Query parameters can be used on either GET, PATCH, PUT or DELETE requests.
+
 ===============================             ===========
 url parameter                               description
 ===============================             ===========
@@ -23,13 +25,15 @@ Additional parameters are available when using an ElasticSearch-enabled collecti
 ===============================             ===========
 url parameter                               description
 ===============================             ===========
-``<field_name>=<keywords>``                 to filter a collection using full-text search on <field_name>, ElasticSearch operators [#]_ can be used, e.g. ``?title=foo AND bar``
+``<field_name>=<keywords>``                 to filter a collection using full-text search on <field_name>, ES operators [#]_ can be used, e.g. ``?title=foo AND bar``
 ``q=<keywords>``                            to filter a collection using full-text search on all fields
 ``_search_fields=<field_list>``             use with ``?q=<keywords>`` to restrict search to specific fields
+``_refresh_index=true``                     to refresh the ES index after performing the operation [#]_ set ``elasticsearch.enable_refresh_query = true`` in your .ini file to enable that feature
 ===============================             ===========
 
 .. [#] To update listfields and dictfields, you can use the following syntax: ``_m=PATCH&<listfield>=<comma_separated_list>&<dictfield>.<key>=<value>``
-.. [#] The full syntax of ElasticSearch querying is beyond the scope of this documentation. You can read more on the ElasticSearch Query String Query `documentation <http://www.elastic.co/guide/en/elasticsearch/reference/1.x/query-dsl-query-string-query.html>`_ to do things like fuzzy search: ``?name=fuzzy~`` or date range search: ``?date=[2015-01-01 TO *]``
+.. [#] The full syntax of ElasticSearch querying is beyond the scope of this documentation. You can read more on the `ElasticSearch Query String Query documentation <http://www.elastic.co/guide/en/elasticsearch/reference/1.x/query-dsl-query-string-query.html>`_ to do things like fuzzy search: ``?name=fuzzy~`` or date range search: ``?date=[2015-01-01 TO *]``
+.. [#] This parameter only works with POST, PATCH, PUT and DELETE methods. Read more on `ElasticSearch Bulk API documentation <https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html#bulk-refresh>`_
 
 update_many()
 -------------

nefertari/elasticsearch.py

@@ -1,4 +1,5 @@
 from __future__ import absolute_import
+import json
 import logging
 from functools import partial
 
@@ -29,15 +30,31 @@ class IndexNotFoundException(Exception):
 
 
 class ESHttpConnection(elasticsearch.Urllib3HttpConnection):
+    def _catch_index_error(self, response):
+        """ Catch and raise index errors which are not critical and thus
+        not raised by elasticsearch-py.
+        """
+        code, headers, raw_data = response
+        if not raw_data:
+            return
+        data = json.loads(raw_data)
+        if not data or not data.get('errors'):
+            return
+        try:
+            error_dict = data['items'][0]['index']
+            message = error_dict['error']
+        except (KeyError, IndexError):
+            return
+        raise exception_response(400, detail=message)
+
     def perform_request(self, *args, **kw):
         try:
             if log.level == logging.DEBUG:
                 msg = str(args)
                 if len(msg) > 512:
                     msg = msg[:300] + '...TRUNCATED...' + msg[-212:]
                 log.debug(msg)
-
-            return super(ESHttpConnection, self).perform_request(*args, **kw)
+            resp = super(ESHttpConnection, self).perform_request(*args, **kw)
         except Exception as e:
             log.error(e.error)
             status_code = e.status_code
@@ -49,6 +66,9 @@ def perform_request(self, *args, **kw):
                 status_code,
                 detail='elasticsearch error.',
                 extra=dict(data=e))
+        else:
+            self._catch_index_error(resp)
+            return resp
 
 
 def includeme(config):
@@ -169,6 +189,13 @@ def setup(cls, settings):
             raise Exception(
                 'Bad or missing settings for elasticsearch. %s' % e)
 
+    def __init__(self, source='', index_name=None, chunk_size=None):
+        self.doc_type = self.src2type(source)
+        self.index_name = index_name or ES.settings.index_name
+        if chunk_size is None:
+            chunk_size = ES.settings.asint('chunk_size')
+        self.chunk_size = chunk_size
+
     @classmethod
     def create_index(cls, index_name=None):
         index_name = index_name or ES.settings.index_name
@@ -188,12 +215,11 @@ def setup_mappings(cls):
         except JHTTPBadRequest as ex:
             raise Exception(ex.json['extra']['data'])
 
-    def __init__(self, source='', index_name=None, chunk_size=None):
-        self.doc_type = self.src2type(source)
-        self.index_name = index_name or ES.settings.index_name
-        if chunk_size is None:
-            chunk_size = ES.settings.asint('chunk_size')
-        self.chunk_size = chunk_size
+    def delete_mapping(self):
+        ES.api.indices.delete_mapping(
+            index=self.index_name,
+            doc_type=self.doc_type,
+        )
 
     def put_mapping(self, body, **kwargs):
         ES.api.indices.put_mapping(

nefertari/scripts/es.py

@@ -39,7 +39,8 @@ def __init__(self, argv, log):
             default=False)
         parser.add_argument(
             '--models',
-            help='Comma-separeted list of model names to index',
+            help=('Comma-separated list of model names to index '
+                  '(required)'),
             required=True)
         parser.add_argument(
             '--params', help='Url-encoded params for each model')
@@ -51,8 +52,9 @@ def __init__(self, argv, log):
             type=int)
         parser.add_argument(
             '--force',
-            help=('Force reindexing of all documents. By default, only '
-                  'documents that are missing from index are indexed.'),
+            help=('Recreate ES mappings and reindex all documents of provided '
+                  'models. By default, only documents that are missing from '
+                  'index are indexed.'),
             action='store_true',
             default=False)
 
@@ -80,6 +82,7 @@ def run(self):
         model_names = split_strip(self.options.models)
 
         for model_name in model_names:
+            self.log.info('Processing model `{}`'.format(model_name))
             model = engine.get_document_cls(model_name)
 
             params = self.options.params or ''
@@ -95,8 +98,16 @@ def run(self):
             documents = to_dicts(query_set)
 
             if self.options.force:
-                es.index(documents)
+                self.log.info('Recreating `{}` ES mapping'.format(model_name))
+                es.delete_mapping()
+                es.put_mapping(body=model.get_es_mapping())
+                self.log.info('Indexing all `{}` documents'.format(
+                    model_name))
+                es.index(documents, refresh_index=self.options.refresh)
             else:
-                es.index_missing_documents(documents)
+                self.log.info('Indexing missing `{}` documents'.format(
+                    model_name))
+                es.index_missing_documents(
+                    documents, refresh_index=self.options.refresh)
 
         return 0

tests/test_elasticsearch.py

@@ -11,8 +11,9 @@
 
 class TestESHttpConnection(object):
 
+    @patch('nefertari.elasticsearch.ESHttpConnection._catch_index_error')
     @patch('nefertari.elasticsearch.log')
-    def test_perform_request_debug(self, mock_log):
+    def test_perform_request_debug(self, mock_log, mock_catch):
         mock_log.level = logging.DEBUG
         conn = es.ESHttpConnection()
         conn.pool = Mock()
@@ -21,8 +22,46 @@ def test_perform_request_debug(self, mock_log):
         mock_log.debug.assert_called_once_with(
             "('POST', 'http://localhost:9200')")
         conn.perform_request('POST', 'http://localhost:9200'*200)
+        assert mock_catch.called
         assert mock_log.debug.call_count == 2
 
+    def test_catch_index_error_no_data(self):
+        conn = es.ESHttpConnection()
+        try:
+            conn._catch_index_error((1, 2, None))
+        except:
+            raise Exception('Unexpected exeption')
+
+    def test_catch_index_error_no_data_loaded(self):
+        conn = es.ESHttpConnection()
+        try:
+            conn._catch_index_error((1, 2, '[]'))
+        except:
+            raise Exception('Unexpected exeption')
+
+    def test_catch_index_error_no_errors(self):
+        conn = es.ESHttpConnection()
+        try:
+            conn._catch_index_error((1, 2, '{"errors":false}'))
+        except:
+            raise Exception('Unexpected exeption')
+
+    def test_catch_index_error_not_index_error(self):
+        conn = es.ESHttpConnection()
+        try:
+            conn._catch_index_error((
+                1, 2,
+                '{"errors":true, "items": [{"foo": "bar"}]}'))
+        except:
+            raise Exception('Unexpected exeption')
+
+    def test_catch_index_error(self):
+        conn = es.ESHttpConnection()
+        with pytest.raises(JHTTPBadRequest):
+            conn._catch_index_error((
+                1, 2,
+                '{"errors":true, "items": [{"index": {"error": "FOO"}}]}'))
+
     def test_perform_request_exception(self):
         conn = es.ESHttpConnection()
         conn.pool = Mock()