From 9e4d0dd25d23c7871b9193db85a24664a7a4c807 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Kr=C3=A1l?= Date: Wed, 30 Sep 2015 19:16:21 +0200 Subject: [PATCH] Move the bulk body construction into _chunk_actions --- elasticsearch/helpers/__init__.py | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/elasticsearch/helpers/__init__.py b/elasticsearch/helpers/__init__.py index f9a822eb..80e93568 100644 --- a/elasticsearch/helpers/__init__.py +++ b/elasticsearch/helpers/__init__.py @@ -40,7 +40,16 @@ def expand_action(data): def _chunk_actions(actions, chunk_size): while True: - yield islice(actions, chunk_size) + bulk_actions = [] + for action, data in islice(actions, chunk_size): + bulk_actions.append(action) + if data is not None: + bulk_actions.append(data) + + if not bulk_actions: + return + + yield bulk_actions def streaming_bulk(client, actions, chunk_size=500, raise_on_error=True, expand_action_callback=expand_action, raise_on_exception=True, @@ -103,16 +112,8 @@ def streaming_bulk(client, actions, chunk_size=500, raise_on_error=True, # if raise on error is set, we need to collect errors per chunk before raising them errors = [] - for chunk in _chunk_actions(actions, chunk_size): + for bulk_actions in _chunk_actions(actions, chunk_size): - bulk_actions = [] - for action, data in chunk: - bulk_actions.append(action) - if data is not None: - bulk_actions.append(data) - - if not bulk_actions: - return try: # send the actual request