Only collect info on failed docs when using the bulk_index helper

elasticsearch/helpers.py

@@ -27,18 +27,21 @@ def bulk_index(client, docs, chunk_size=500, stats_only=False, **kwargs):
     :arg docs: iterator containing the docs
     :arg chunk_size: number of docs in one chunk sent to es (default: 500)
     :arg stats_only: if `True` only report number of successful/failed
-        operations
+        operations instead of just number of successful and a list of error
+        responses
 
     Any additional keyword arguments will be passed to the bulk API itself.
     """
-    if stats_only:
-        success, failed = 0, 0
-    else:
-        success, failed = [], []
+    success, failed = 0, 0
+
+    # list of errors to be collected when 
+    errors = []
+
     docs = iter(docs)
     while True:
         chunk = islice(docs, chunk_size)
         bulk_actions = []
+        ndocs = 0
         for d in chunk:
             action = {'index': {}}
             for key in ('_index', '_parent', '_percolate', '_routing',
@@ -48,25 +51,23 @@ def bulk_index(client, docs, chunk_size=500, stats_only=False, **kwargs):
 
             bulk_actions.append(action)
             bulk_actions.append(d.get('_source', d))
+            ndocs += 1
 
         if not bulk_actions:
-            return success, failed
+            return success, failed if stats_only else errors
 
         resp = client.bulk(bulk_actions, **kwargs)
 
         for req, item in zip(bulk_actions[::2], resp['items']):
             # TODO: better reporting
             ok = item['index' if '_id' in req['index'] else 'create'].get('ok')
-            if stats_only:
-                if ok:
-                    success += 1
-                else:
+            if not ok:
+                if stats_only:
                     failed += 1
-            else:
-                if ok:
-                    success.append(item)
                 else:
-                    failed.append(item)
+                    errors.append(item)
+            else:
+                success += 1
 
 def scan(client, query=None, scroll='5m', **kwargs):
     """

test_elasticsearch/test_server/test_helpers.py

@@ -7,7 +7,7 @@ class TestBulkIndex(ElasticTestCase):
         docs = [{"answer": x, '_id': x} for x in range(100)]
         success, failed = helpers.bulk_index(self.client, docs, index='test-index', doc_type='answers', refresh=True)
 
-        self.assertEquals(100, len(success))
+        self.assertEquals(100, success)
         self.assertFalse(failed)
         self.assertEquals(100, self.client.count(index='test-index', doc_type='answers')['count'])
         self.assertEquals({"answer": 42}, self.client.get(index='test-index', doc_type='answers', id=42)['_source'])
@@ -20,6 +20,28 @@ class TestBulkIndex(ElasticTestCase):
         self.assertEquals(0, failed)
         self.assertEquals(100, self.client.count(index='test-index', doc_type='answers')['count'])
 
+    def test_errors_are_reported_correctly(self):
+        self.client.indices.create("i",
+            {
+                "mappings": {"t": {"properties": {"a": {"type": "integer"}}}},
+                "settings": {"number_of_shards": 1, "number_of_replicas": 0}
+            })
+        self.client.cluster.health(wait_for_status="yellow")
+
+        success, failed = helpers.bulk_index(
+            self.client,
+            [{"a": 42}, {"a": "c", '_id': 42}],
+            index="i",
+            doc_type="t"
+        )
+        self.assertEquals(1, success)
+        self.assertEquals(1, len(failed))
+        error = failed[0]
+        self.assertEquals('42', error['index']['_id'])
+        self.assertEquals('t', error['index']['_type'])
+        self.assertEquals('i', error['index']['_index'])
+        self.assertIn('MapperParsingException', error['index']['error'])
+
     def test_errors_are_collected_properly(self):
         self.client.indices.create("i",
             {