diff --git a/elasticsearch/helpers.py b/elasticsearch/helpers.py index 802dce9c..d217fc60 100644 --- a/elasticsearch/helpers.py +++ b/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): """ diff --git a/test_elasticsearch/test_server/test_helpers.py b/test_elasticsearch/test_server/test_helpers.py index 967e74f6..b2f36d96 100644 --- a/test_elasticsearch/test_server/test_helpers.py +++ b/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", {