Add raise_on_error parameter to bulk_index to enable short-circuiting bulk loading

elasticsearch/helpers.py

@@ -1,6 +1,10 @@
 from itertools import islice
 
-def bulk_index(client, docs, chunk_size=500, stats_only=False, **kwargs):
+from elasticsearch.exceptions import ElasticsearchException
+
+class BulkIndexError(ElasticsearchException): pass
+
+def bulk_index(client, docs, chunk_size=500, stats_only=False, raise_on_error=False, **kwargs):
     """
     Helper for the :meth:`~elasticsearch.Elasticsearch.bulk` api that provides
     a more human friendly interface - it consumes an iterator of documents and
@@ -27,8 +31,9 @@ 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 instead of just number of successful and a list of error
+        operations instead of just number of successful and a list of error responses
-        responses
+    :arg raise_on_error: raise `BulkIndexError` if some documents failed to
+        index (and stop sending chunks to the server)
 
     Any additional keyword arguments will be passed to the bulk API itself.
     """
@@ -59,7 +64,6 @@ def bulk_index(client, docs, chunk_size=500, stats_only=False, **kwargs):
         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 not ok:
                 if stats_only:
@@ -69,6 +73,9 @@ def bulk_index(client, docs, chunk_size=500, stats_only=False, **kwargs):
             else:
                 success += 1
 
+        if (failed or errors) and raise_on_error:
+            raise BulkIndexError(failed, errors)
+
 def scan(client, query=None, scroll='5m', **kwargs):
     """
     Simple abstraction on top of the

test_elasticsearch/test_server/test_helpers.py

@@ -42,6 +42,22 @@ class TestBulkIndex(ElasticTestCase):
         self.assertEquals('i', error['index']['_index'])
         self.assertIn('MapperParsingException', error['index']['error'])
 
+    def test_error_is_raised_if_requested(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")
+
+        self.assertRaises(helpers.BulkIndexError, helpers.bulk_index,
+            self.client,
+            [{"a": 42}, {"a": "c"}],
+            index="i",
+            doc_type="t",
+            raise_on_error=True
+        )
+
     def test_errors_are_collected_properly(self):
         self.client.indices.create("i",
             {