Add deduplication and test

Author: Pijukatel

src/apify/storage_clients/_apify/_request_queue_client.py

@@ -238,17 +238,48 @@ async def add_batch_of_requests(
         Returns:
             Response containing information about the added requests.
         """
-        # Prepare requests for API by converting to dictionaries.
-        requests_dict = [
-            request.model_dump(
-                by_alias=True,
-                exclude={'id'},  # Exclude ID fields from requests since the API doesn't accept them.
-            )
-            for request in requests
-        ]
+        # Do not try to add previously added requests to avoid pointless expensive calls to API
+        new_requests: list[Request] = []
+        already_present_requests: list[ProcessedRequest] = []
+        for request in requests:
+            if self._requests_cache.get(request.id):
+                # We are no sure if it was already handled at this point, and it is not worth calling API for it.
+                already_present_requests.append(
+                    {
+                        'id': request.id,
+                        'uniqueKey': request.unique_key,
+                        'wasAlreadyPresent': True,
+                        'wasAlreadyHandled': request.was_already_handled,
+                    }
+                )
 
-        # Send requests to API.
-        response = await self._api_client.batch_add_requests(requests=requests_dict, forefront=forefront)
+            else:
+                new_requests.append(request)
+
+        if new_requests:
+            # Prepare requests for API by converting to dictionaries.
+            requests_dict = [
+                request.model_dump(
+                    by_alias=True,
+                    exclude={'id'},  # Exclude ID fields from requests since the API doesn't accept them.
+                )
+                for request in new_requests
+            ]
+
+            # Send requests to API.
+            response = await self._api_client.batch_add_requests(requests=requests_dict, forefront=forefront)
+            # Add new requests to the cache.
+            for processed_request_raw in response['processedRequests']:
+                processed_request = ProcessedRequest.model_validate(processed_request_raw)
+                self._cache_request(
+                    unique_key_to_request_id(processed_request.unique_key),
+                    processed_request,
+                    forefront=False,
+                )
+            # Add the locally known already present processed requests based on the local cache.
+            response['processedRequests'].extend(already_present_requests)
+        else:
+            response = {'unprocessedRequests': [], 'processedRequests': already_present_requests}
 
         # Update assumed total count for newly added requests.
         api_response = AddRequestsResponse.model_validate(response)

tests/integration/test_actor_request_queue.py

@@ -110,3 +110,42 @@ async def test_request_queue_is_finished(
 
         await request_queue.mark_request_as_handled(request)
         assert await request_queue.is_finished()
+
+
+async def test_request_queue_deduplication(
+    make_actor: MakeActorFunction,
+    run_actor: RunActorFunction,
+) -> None:
+    """Test that the deduplication works correctly. Try to add 2 same requests, but it should call API just once.
+
+    This tests internal optimization that changes no behavior for the user.
+    The functions input/output behave the same way,it only uses less amount of API calls.
+    """
+
+    async def main() -> None:
+        import asyncio
+
+        from apify import Actor, Request
+
+        async with Actor:
+            request = Request.from_url('http://example.com')
+            rq = await Actor.open_request_queue(name='test-deduplication', force_cloud=True)
+
+            await asyncio.sleep(10)  # Wait to be sure that metadata are updated
+            stats_before = (await Actor.apify_client.request_queue(request_queue_id=rq.id).get()).get('stats', {})
+            Actor.log.info(stats_before)
+
+            # Add same request twice
+            await rq.add_request(request)
+            await rq.add_request(request)
+
+            await asyncio.sleep(10)  # Wait to be sure that metadata are updated
+            stats_after = (await Actor.apify_client.request_queue(request_queue_id=rq.id).get()).get('stats', {})
+            Actor.log.info(stats_after)
+
+            assert (stats_after['writeCount'] - stats_before['writeCount']) ==  1
+
+    actor = await make_actor(label='rq-deduplication', main_func=main)
+    run_result = await run_actor(actor)
+
+    assert run_result.status == 'SUCCEEDED'