diff --git a/elasticsearch/helpers.py b/elasticsearch/helpers.py index d217fc60..6577849e 100644 --- a/elasticsearch/helpers.py +++ b/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 - responses + operations instead of just number of successful and a list of error 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 diff --git a/test_elasticsearch/test_server/test_helpers.py b/test_elasticsearch/test_server/test_helpers.py index b2f36d96..31a2f970 100644 --- a/test_elasticsearch/test_server/test_helpers.py +++ b/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", {