diff --git a/Changelog.rst b/Changelog.rst index b90eafe0..3afc0bff 100644 --- a/Changelog.rst +++ b/Changelog.rst @@ -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) ------------------ diff --git a/elasticsearch/helpers/__init__.py b/elasticsearch/helpers/__init__.py index c1c6528a..cff3795c 100644 --- a/elasticsearch/helpers/__init__.py +++ b/elasticsearch/helpers/__init__.py @@ -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) diff --git a/test_elasticsearch/test_server/test_helpers.py b/test_elasticsearch/test_server/test_helpers.py index ea141085..b76639f2 100644 --- a/test_elasticsearch/test_server/test_helpers.py +++ b/test_elasticsearch/test_server/test_helpers.py @@ -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()