Add option to pass kwargs to scan and bulk in reindex helper

This commit is contained in:
Honza Král
2015-01-30 00:04:04 +01:00
parent d9acd16850
commit f065817e89
3 changed files with 20 additions and 3 deletions
+2
View File
@@ -7,6 +7,8 @@ Changelog
-----------
* Using insecure SSL configuration (``verify_cert=False``) raises a warning
* enable ``reindex`` helper to accept any kwargs for underlying ``bulk`` and
``scan`` calls
1.3.0 (2014-12-31)
------------------
+9 -3
View File
@@ -236,7 +236,9 @@ def scan(client, query=None, scroll='5m', preserve_order=False, **kwargs):
if scroll_id is None:
break
def reindex(client, source_index, target_index, query=None, target_client=None, chunk_size=500, scroll='5m'):
def reindex(client, source_index, target_index, query=None, target_client=None,
chunk_size=500, scroll='5m', scan_kwargs={}, bulk_kwargs={}):
"""
Reindex all documents from one index that satisfy a given query
to another, potentially (if `target_client` is specified) on a different cluster.
@@ -256,14 +258,18 @@ def reindex(client, source_index, target_index, query=None, target_client=None,
:arg chunk_size: number of docs in one chunk sent to es (default: 500)
:arg scroll: Specify how long a consistent view of the index should be
maintained for scrolled search
:arg scan_kwargs: additional kwargs to be passed to
:func:`~elasticsearch.helpers.scan`
:arg bulk_kwargs: additional kwargs to be passed to
:func:`~elasticsearch.helpers.bulk`
"""
target_client = client if target_client is None else target_client
docs = scan(client, query=query, index=source_index, scroll=scroll)
docs = scan(client, query=query, index=source_index, scroll=scroll, **scan_kwargs)
def _change_doc_index(hits, index):
for h in hits:
h['_index'] = index
yield h
return bulk(target_client, _change_doc_index(docs, target_index),
chunk_size=chunk_size, stats_only=True)
chunk_size=chunk_size, stats_only=True, **bulk_kwargs)
@@ -215,6 +215,15 @@ class TestReindex(ElasticsearchTestCase):
bulk.append({"answer": x, "correct": x == 42})
self.client.bulk(bulk, refresh=True)
def test_reindex_passes_kwargs_to_scan_and_bulk(self):
helpers.reindex(self.client, "test_index", "prod_index", scan_kwargs={'doc_type': 'answers'}, bulk_kwargs={'refresh': True})
self.assertTrue(self.client.indices.exists("prod_index"))
self.assertFalse(self.client.indices.exists_type(index='prod_index', doc_type='questions'))
self.assertEquals(50, self.client.count(index='prod_index', doc_type='answers')['count'])
self.assertEquals({"answer": 42, "correct": True}, self.client.get(index="prod_index", doc_type="answers", id=42)['_source'])
def test_reindex_accepts_a_query(self):
helpers.reindex(self.client, "test_index", "prod_index", query={"query": {"filtered": {"filter": {"term": {"_type": "answers"}}}}})
self.client.indices.refresh()