actions/opensearch-pyd
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Don't keep the data twice in bulk helper

Browse Source
This commit is contained in:
Honza Král
2015-09-30 19:03:49 +02:00
parent 0ad6b94d1e
commit a177375c25
1 changed files with 13 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
elasticsearch/helpers/__init__.py
+13 -9
View File
@@ -105,10 +105,6 @@ def streaming_bulk(client, actions, chunk_size=500, raise_on_error=True,
for chunk in _chunk_actions(actions, chunk_size): 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 = [] bulk_actions = []
for action, data in chunk: for action, data in chunk:
bulk_actions.append(action) 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 # if we are not propagating, mark all actions in current chunk as failed
err_message = str(e) err_message = str(e)
exc_errors = [] exc_errors = []
for action, data in chunk: bulk_data = iter(bulk_actions)
info = {"error": err_message, "status": e.status_code, "exception": e, "data": data} while True:
op_type, action = action.popitem() try:
info.update(action) # collect all the information about failed actions
exc_errors.append({op_type: info}) 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 # emulate standard behavior for failed actions
if raise_on_error: if raise_on_error: