[7.16] Always set 'sort' and 'scroll' for scan helper

Author: sethmlarson

elasticsearch/_async/helpers.py

@@ -26,6 +26,7 @@
 from ..exceptions import NotFoundError, TransportError
 from ..helpers.actions import (
     _ActionChunker,
+    _add_helper_meta_to_kwargs,
     _process_bulk_chunk_error,
     _process_bulk_chunk_success,
     expand_action,
@@ -332,7 +333,9 @@ async def async_scan(
         )
 
     """
-    scroll_kwargs = scroll_kwargs or {}
+    scroll_kwargs = scroll_kwargs.copy() if scroll_kwargs else {}
+    scroll_kwargs["scroll"] = scroll
+    _add_helper_meta_to_kwargs(scroll_kwargs, "s")
 
     if not preserve_order:
         query = query.copy() if query else {}
@@ -356,9 +359,11 @@ async def async_scan(
     search_kwargs = kwargs.copy()
     if query:
         search_kwargs.update(query)
-    resp = await client.search(
-        scroll=scroll, size=size, request_timeout=request_timeout, **search_kwargs
-    )
+    search_kwargs["scroll"] = scroll
+    search_kwargs["size"] = size
+    search_kwargs["request_timeout"] = request_timeout
+    _add_helper_meta_to_kwargs(search_kwargs, "s")
+    resp = await client.search(**search_kwargs)
     scroll_id = resp.get("_scroll_id")
 
     try:
@@ -390,9 +395,9 @@ async def async_scan(
                             shards_total,
                         ),
                     )
-            resp = await client.scroll(
-                scroll_id=scroll_id, scroll=scroll, **scroll_kwargs
-            )
+
+            scroll_kwargs["scroll_id"] = scroll_id
+            resp = await client.scroll(**scroll_kwargs)
             scroll_id = resp.get("_scroll_id")
 
     finally:

elasticsearch/helpers/actions.py

@@ -543,7 +543,8 @@ def scan(
         )
 
     """
-    scroll_kwargs = scroll_kwargs or {}
+    scroll_kwargs = scroll_kwargs.copy() if scroll_kwargs else {}
+    scroll_kwargs["scroll"] = scroll
     _add_helper_meta_to_kwargs(scroll_kwargs, "s")
 
     if not preserve_order:
@@ -568,9 +569,11 @@ def scan(
     search_kwargs = kwargs.copy()
     if query:
         search_kwargs.update(query)
-    resp = client.search(
-        scroll=scroll, size=size, request_timeout=request_timeout, **search_kwargs
-    )
+    search_kwargs["scroll"] = scroll
+    search_kwargs["size"] = size
+    search_kwargs["request_timeout"] = request_timeout
+    _add_helper_meta_to_kwargs(search_kwargs, "s")
+    resp = client.search(**search_kwargs)
     scroll_id = resp.get("_scroll_id")
 
     try:
@@ -602,7 +605,9 @@ def scan(
                             shards_total,
                         ),
                     )
-            resp = client.scroll(scroll_id=scroll_id, scroll=scroll, **scroll_kwargs)
+
+            scroll_kwargs["scroll_id"] = scroll_id
+            resp = client.scroll(**scroll_kwargs)
             scroll_id = resp.get("_scroll_id")
 
     finally:

test_elasticsearch/test_async/test_server/test_helpers.py

@@ -756,6 +756,59 @@ async def test_scan_auth_kwargs_favor_scroll_kwargs_option(
                     }
                     assert async_client.scroll.call_args[1]["sort"] == "asc"
 
+    async def test_scan_duplicate_parameters(self, async_client):
+        with patch.object(async_client, "search") as search_mock, patch.object(
+            async_client, "scroll"
+        ) as scroll_mock, patch.object(
+            async_client, "clear_scroll"
+        ) as clear_scroll_mock:
+            search_mock.return_value = MockResponse(
+                {
+                    "_scroll_id": "scroll_id",
+                    "_shards": {"successful": 5, "total": 5, "skipped": 0},
+                    "hits": {"hits": [{"field": "value"}]},
+                }
+            )
+            scroll_mock.return_value = MockResponse(
+                {
+                    "_scroll_id": "scroll_id",
+                    "_shards": {"successful": 5, "total": 5, "skipped": 0},
+                    "hits": {"hits": []},
+                }
+            )
+            clear_scroll_mock.return_value = MockResponse({"acknowledged": True})
+            data = [
+                x
+                async for x in helpers.async_scan(
+                    async_client,
+                    index="test_index",
+                    size=10,
+                    query={"size": 1},
+                    scroll_kwargs={"scroll": "10m", "rest_total_hits_as_int": True},
+                )
+            ]
+
+            assert data == [{"field": "value"}]
+            search_mock.assert_called_with(
+                index="test_index",
+                size=10,
+                sort="_doc",
+                scroll="5m",
+                request_timeout=None,
+                params={"__elastic_client_meta": (("h", "s"),)},
+            )
+            scroll_mock.assert_called_with(
+                scroll="5m",
+                rest_total_hits_as_int=True,
+                params={"__elastic_client_meta": (("h", "s"),)},
+                scroll_id="scroll_id",
+            )
+            clear_scroll_mock.assert_called_with(
+                scroll_id="scroll_id",
+                ignore=(404,),
+                params={"__elastic_client_meta": (("h", "s"),)},
+            )
+
 
 @pytest.fixture(scope="function")
 async def reindex_setup(async_client):

test_elasticsearch/test_server/test_helpers.py

@@ -568,6 +568,57 @@ def test_scan_auth_kwargs_favor_scroll_kwargs_option(self):
             )
             self.assertEqual(client_mock.scroll.call_args[1]["sort"], "asc")
 
+    def test_scan_duplicate_parameters(self):
+        with patch.object(self.client, "search") as search_mock, patch.object(
+            self.client, "scroll"
+        ) as scroll_mock, patch.object(
+            self.client, "clear_scroll"
+        ) as clear_scroll_mock:
+            search_mock.return_value = {
+                "_scroll_id": "scroll_id",
+                "_shards": {"successful": 5, "total": 5, "skipped": 0},
+                "hits": {"hits": [{"field": "value"}]},
+            }
+
+            scroll_mock.return_value = {
+                "_scroll_id": "scroll_id",
+                "_shards": {"successful": 5, "total": 5, "skipped": 0},
+                "hits": {"hits": []},
+            }
+
+            clear_scroll_mock.return_value = {"acknowledged": True}
+            data = [
+                x
+                for x in helpers.scan(
+                    self.client,
+                    index="test_index",
+                    size=10,
+                    query={"size": 1},
+                    scroll_kwargs={"scroll": "10m", "rest_total_hits_as_int": True},
+                )
+            ]
+
+            assert data == [{"field": "value"}]
+            search_mock.assert_called_with(
+                index="test_index",
+                size=10,
+                sort="_doc",
+                scroll="5m",
+                request_timeout=None,
+                params={"__elastic_client_meta": (("h", "s"),)},
+            )
+            scroll_mock.assert_called_with(
+                scroll="5m",
+                rest_total_hits_as_int=True,
+                params={"__elastic_client_meta": (("h", "s"),)},
+                scroll_id="scroll_id",
+            )
+            clear_scroll_mock.assert_called_with(
+                scroll_id="scroll_id",
+                ignore=(404,),
+                params={"__elastic_client_meta": (("h", "s"),)},
+            )
+
     @patch("elasticsearch.helpers.actions.logger")
     def test_logger(self, logger_mock):
         bulk = []