From 0ad6b94d1e08e37d6b654a96f6121a46dbc105fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Kr=C3=A1l?= Date: Wed, 30 Sep 2015 18:23:42 +0200 Subject: [PATCH] Extract chunking from streaming_bulk --- elasticsearch/helpers/__init__.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/elasticsearch/helpers/__init__.py b/elasticsearch/helpers/__init__.py index 1200efc4..ff318e7f 100644 --- a/elasticsearch/helpers/__init__.py +++ b/elasticsearch/helpers/__init__.py @@ -38,6 +38,9 @@ def expand_action(data): return action, data.get('_source', data) +def _chunk_actions(actions, chunk_size): + while True: + yield islice(actions, chunk_size) def streaming_bulk(client, actions, chunk_size=500, raise_on_error=True, expand_action_callback=expand_action, raise_on_exception=True, @@ -100,8 +103,7 @@ 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 = [] - while True: - chunk = islice(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: