From 17c4a3dd213adfe8a4d006e18a33d8b9c10ef32a Mon Sep 17 00:00:00 2001
From: Honza Kral <honza.kral@gmail.com>
Date: Mon, 30 Sep 2013 19:05:12 +0200
Subject: [PATCH] stats_only parameter to bulk_index helper for large loads`

---
 elasticsearch/helpers.py                      | 24 ++++++++++++++-----
 .../test_server/test_helpers.py               |  8 +++++++
 2 files changed, 26 insertions(+), 6 deletions(-)

diff --git a/elasticsearch/helpers.py b/elasticsearch/helpers.py
index 51946872..1274ab27 100644
--- a/elasticsearch/helpers.py
+++ b/elasticsearch/helpers.py
@@ -1,6 +1,6 @@
 from itertools import islice
 
-def bulk_index(client, docs, chunk_size=500, **kwargs):
+def bulk_index(client, docs, chunk_size=500, stats_only=False, **kwargs):
     """
     Helper for the :meth:`~elasticsearch.Elasticsearch.bulk` api that provides
     a more human friendly interface - it consumes an iterator of documents and
@@ -25,10 +25,15 @@ def bulk_index(client, docs, chunk_size=500, **kwargs):
     :arg client: instance of :class:`~elasticsearch.Elasticsearch` to use
     :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
 
     Any additional keyword arguments will be passed to the bulk API itself.
     """
-    success, failed = [], []
+    if stats_only:
+        success, failed = 0, 0
+    else:
+        success, failed = [], []
     docs = iter(docs)
     while True:
         chunk = islice(docs, chunk_size)
@@ -49,10 +54,16 @@ def bulk_index(client, docs, chunk_size=500, **kwargs):
         resp = client.bulk(bulk_actions, **kwargs)
 
         for item in resp['items']:
-            if item['create']['ok']:
-                success.append(item)
+            if stats_only:
+                if item['create']['ok']:
+                    success += 1
+                else:
+                    failed += 1
             else:
-                failed.append(item)
+                if item['create']['ok']:
+                    success.append(item)
+                else:
+                    failed.append(item)
 
 def scan(client, query=None, scroll='5m', **kwargs):
     """
@@ -108,4 +119,5 @@ def reindex(client, source_index, target_index, target_client=None, chunk_size=5
             h['_index'] = index
             yield h
 
-    return bulk_index(target_client, _change_doc_index(docs, target_index), chunk_size=chunk_size)
+    return bulk_index(target_client, _change_doc_index(docs, target_index),
+        chunk_size=chunk_size, stats_only=True)
diff --git a/test_elasticsearch/test_server/test_helpers.py b/test_elasticsearch/test_server/test_helpers.py
index 6ff7a1b2..61a7980d 100644
--- a/test_elasticsearch/test_server/test_helpers.py
+++ b/test_elasticsearch/test_server/test_helpers.py
@@ -11,6 +11,14 @@ class TestBulkIndex(ElasticTestCase):
         self.assertFalse(failed)
         self.assertEquals(100, self.client.count(index='test-index', doc_type='answers')['count'])
 
+    def test_stats_only_reports_numbers(self):
+        docs = [{"answer": x} for x in range(100)]
+        success, failed = helpers.bulk_index(self.client, docs, index='test-index', doc_type='answers', refresh=True, stats_only=True)
+
+        self.assertEquals(100, success)
+        self.assertEquals(0, failed)
+        self.assertEquals(100, self.client.count(index='test-index', doc_type='answers')['count'])
+
 class TestScan(ElasticTestCase):
     def test_all_documents_are_read(self):
         bulk = []