From a177375c25cfc26c0c4c27632d06524ab0fcd3f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Kr=C3=A1l?= Date: Wed, 30 Sep 2015 19:00:49 +0200 Subject: [PATCH] Don't keep the data twice in bulk helper --- elasticsearch/helpers/__init__.py | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/elasticsearch/helpers/__init__.py b/elasticsearch/helpers/__init__.py index ff318e7f..f9a822eb 100644 --- a/elasticsearch/helpers/__init__.py +++ b/elasticsearch/helpers/__init__.py @@ -105,10 +105,6 @@ def streaming_bulk(client, actions, chunk_size=500, raise_on_error=True, for chunk in _chunk_actions(actions, chunk_size): - # raise on exception means we might need to iterate on chunk twice - if not raise_on_exception: - chunk = list(chunk) - bulk_actions = [] for action, data in chunk: bulk_actions.append(action) @@ -129,11 +125,19 @@ def streaming_bulk(client, actions, chunk_size=500, raise_on_error=True, # if we are not propagating, mark all actions in current chunk as failed err_message = str(e) exc_errors = [] - for action, data in chunk: - info = {"error": err_message, "status": e.status_code, "exception": e, "data": data} - op_type, action = action.popitem() - info.update(action) - exc_errors.append({op_type: info}) + bulk_data = iter(bulk_actions) + while True: + try: + # collect all the information about failed actions + action = next(bulk_data) + op_type, action = action.popitem() + info = {"error": err_message, "status": e.status_code, "exception": e} + if op_type != 'delete': + info['data'] = next(bulk_data) + info.update(action) + exc_errors.append({op_type: info}) + except StopIteration: + break # emulate standard behavior for failed actions if raise_on_error: