Do early response validation

Author: Pijukatel

src/apify/storage_clients/_apify/_request_queue_client.py

@@ -245,20 +245,22 @@ async def add_batch_of_requests(
         # Do not try to add previously added requests to avoid pointless expensive calls to API
 
         new_requests: list[Request] = []
-        already_present_requests: list[dict[str, str | bool]] = []
+        already_present_requests: list[ProcessedRequest] = []
 
         for request in requests:
             if self._requests_cache.get(request.id):
                 # We are not sure if it was already handled at this point, and it is not worth calling API for it.
                 # It could have been handled by another client in the meantime, so cached information about
                 # `request.was_already_handled` is not reliable.
                 already_present_requests.append(
-                    {
-                        'id': request.id,
-                        'uniqueKey': request.unique_key,
-                        'wasAlreadyPresent': True,
-                        'wasAlreadyHandled': request.was_already_handled,
-                    }
+                    ProcessedRequest.model_validate(
+                        {
+                            'id': request.id,
+                            'uniqueKey': request.unique_key,
+                            'wasAlreadyPresent': True,
+                            'wasAlreadyHandled': request.was_already_handled,
+                        }
+                    )
                 )
 
             else:
@@ -288,25 +290,29 @@ async def add_batch_of_requests(
             ]
 
             # Send requests to API.
-            response = await self._api_client.batch_add_requests(requests=requests_dict, forefront=forefront)
+            api_response = AddRequestsResponse.model_validate(
+                await self._api_client.batch_add_requests(requests=requests_dict, forefront=forefront)
+            )
+
             # Add the locally known already present processed requests based on the local cache.
-            response['processedRequests'].extend(already_present_requests)
+            api_response.processed_requests.extend(already_present_requests)
 
             # Remove unprocessed requests from the cache
-            for unprocessed in response['unprocessedRequests']:
-                self._requests_cache.pop(unique_key_to_request_id(unprocessed['uniqueKey']), None)
+            for unprocessed_request in api_response.unprocessed_requests:
+                self._requests_cache.pop(unique_key_to_request_id(unprocessed_request.unique_key), None)
 
         else:
-            response = {'unprocessedRequests': [], 'processedRequests': already_present_requests}
+            api_response = AddRequestsResponse.model_validate(
+                {'unprocessedRequests': [], 'processedRequests': already_present_requests}
+            )
 
         logger.debug(
             f'Tried to add new requests: {len(new_requests)}, '
-            f'succeeded to add new requests: {len(response["processedRequests"]) - len(already_present_requests)}, '
+            f'succeeded to add new requests: {len(api_response.processed_requests) - len(already_present_requests)}, '
             f'skipped already present requests: {len(already_present_requests)}'
         )
 
         # Update assumed total count for newly added requests.
-        api_response = AddRequestsResponse.model_validate(response)
         new_request_count = 0
         for processed_request in api_response.processed_requests:
             if not processed_request.was_already_present and not processed_request.was_already_handled:

tests/integration/test_actor_request_queue.py

@@ -273,7 +273,12 @@ async def add_requests_worker() -> None:
 async def test_request_queue_deduplication_unprocessed_requests(
     apify_named_rq: RequestQueue,
 ) -> None:
-    """Test that the deduplication does not add unprocessed requests to the cache."""
+    """Test that the deduplication does not add unprocessed requests to the cache.
+
+    In this test the first call is "hardcoded" to fail, even on all retries, so it never even sends the API request and
+    thus has no chance of increasing the `writeCount`. The second call can increase the `writeCount` only if it is not
+    cached, as cached requests do not make the call (tested in other tests). So this means the `unprocessedRequests`
+    request was intentionally not cached."""
     logging.getLogger('apify.storage_clients._apify._request_queue_client').setLevel(logging.DEBUG)
 
     await asyncio.sleep(10)  # Wait to be sure that metadata are updated