Remove unnecessary complexity

Author: Pijukatel

src/apify_client/clients/resource_clients/request_queue.py

@@ -36,11 +36,6 @@ class BatchAddRequestsResult(TypedDict):
     unprocessedRequests: list[dict]
 
 
-def _get_unprocessed_request_from_request(request: dict[str, str]) -> dict[str, str]:
-    relevant_keys = {'url', 'uniqueKey', 'method'}
-    return {key: value for key, value in request.items() if key in relevant_keys}
-
-
 class RequestQueueClient(ResourceClient):
     """Sub-client for manipulating a single request queue."""
 
@@ -323,14 +318,11 @@ def batch_add_requests(
             queue.put(batch)
 
         processed_requests = list[dict]()
-        unprocessed_requests = dict[str, dict]()
+        unprocessed_requests = list[dict]()
 
         # Process all batches in the queue sequentially.
         while not queue.empty():
             request_batch = queue.get()
-            # All requests are considered unprocessed unless explicitly mentioned in `processedRequests` response.
-            for request in request_batch:
-                unprocessed_requests[request['uniqueKey']] = _get_unprocessed_request_from_request(request)
 
             # Send the batch to the API.
             response = self.http_client.call(
@@ -343,13 +335,11 @@ def batch_add_requests(
 
             response_parsed = parse_date_fields(pluck_data(response.json()))
             processed_requests.extend(response_parsed.get('processedRequests', []))
-
-            for processed_request in response_parsed.get('processedRequests', []):
-                unprocessed_requests.pop(processed_request['uniqueKey'], None)
+            unprocessed_requests.extend(response_parsed.get('unprocessedRequests', []))
 
         return {
             'processedRequests': processed_requests,
-            'unprocessedRequests': list(unprocessed_requests.values()),
+            'unprocessedRequests': unprocessed_requests,
         }
 
     def batch_delete_requests(self, requests: list[dict]) -> dict:
@@ -650,39 +640,36 @@ async def _batch_add_requests_worker(
         Return result containing lists of processed and unprocessed requests by the worker.
         """
         processed_requests = list[dict]()
-        unprocessed_requests = dict[str, dict]()
+        unprocessed_requests = list[dict]()
 
         while True:
             # Get the next batch from the queue.
             try:
                 request_batch = await queue.get()
-                # All requests are considered unprocessed unless explicitly mentioned in `processedRequests` response.
-                for request in request_batch:
-                    unprocessed_requests[request['uniqueKey']] = _get_unprocessed_request_from_request(request)
-
             except asyncio.CancelledError:
                 break
 
-            # Send the batch to the API.
-            response = await self.http_client.call(
-                url=self._url('requests/batch'),
-                method='POST',
-                params=request_params,
-                json=list(request_batch),
-                timeout_secs=_MEDIUM_TIMEOUT,
-            )
-
-            response_parsed = parse_date_fields(pluck_data(response.json()))
-            processed_requests.extend(response_parsed.get('processedRequests', []))
-
-            for processed_request in response_parsed.get('processedRequests', []):
-                unprocessed_requests.pop(processed_request['uniqueKey'], None)
-
-            queue.task_done()
+            try:
+                # Send the batch to the API.
+                response = await self.http_client.call(
+                    url=self._url('requests/batch'),
+                    method='POST',
+                    params=request_params,
+                    json=list(request_batch),
+                    timeout_secs=_MEDIUM_TIMEOUT,
+                )
+
+                response_parsed = parse_date_fields(pluck_data(response.json()))
+                processed_requests.extend(response_parsed.get('processedRequests', []))
+                unprocessed_requests.extend(response_parsed.get('unprocessedRequests', []))
+
+            finally:
+                # Mark the batch as done whether it succeeded or failed.
+                queue.task_done()
 
         return {
             'processedRequests': processed_requests,
-            'unprocessedRequests': list(unprocessed_requests.values()),
+            'unprocessedRequests': unprocessed_requests,
         }
 
     async def batch_add_requests(

tests/unit/test_client_request_queue.py

@@ -1,5 +1,7 @@
+import pytest
 import respx
 
+import apify_client
 from apify_client import ApifyClient, ApifyClientAsync
 
 _PARTIALLY_ADDED_BATCH_RESPONSE_CONTENT = """{
@@ -24,8 +26,8 @@
 
 
 @respx.mock
-async def test_batch_not_processed_due_to_exception_async() -> None:
-    """Test that all requests are unprocessed unless explicitly stated by the server that they have been processed."""
+async def test_batch_not_processed_raises_exception_async() -> None:
+    """Test that client exceptions are not silently ignored"""
     client = ApifyClientAsync(token='')
 
     respx.route(method='POST', host='api.apify.com').mock(return_value=respx.MockResponse(401))
@@ -35,8 +37,8 @@ async def test_batch_not_processed_due_to_exception_async() -> None:
     ]
     rq_client = client.request_queue(request_queue_id='whatever')
 
-    response = await rq_client.batch_add_requests(requests=requests)
-    assert response['unprocessedRequests'] == requests
+    with pytest.raises(apify_client._errors.ApifyApiError):
+        await rq_client.batch_add_requests(requests=requests)
 
 
 @respx.mock
@@ -58,8 +60,8 @@ async def test_batch_processed_partially_async() -> None:
 
 
 @respx.mock
-def test_batch_not_processed_due_to_exception_sync() -> None:
-    """Test that all requests are unprocessed unless explicitly stated by the server that they have been processed."""
+def test_batch_not_processed_raises_exception_sync() -> None:
+    """Test that client exceptions are not silently ignored"""
     client = ApifyClient(token='')
 
     respx.route(method='POST', host='api.apify.com').mock(return_value=respx.MockResponse(401))
@@ -69,8 +71,8 @@ def test_batch_not_processed_due_to_exception_sync() -> None:
     ]
     rq_client = client.request_queue(request_queue_id='whatever')
 
-    response = rq_client.batch_add_requests(requests=requests)
-    assert response['unprocessedRequests'] == requests
+    with pytest.raises(apify_client._errors.ApifyApiError):
+        rq_client.batch_add_requests(requests=requests)
 
 
 @respx.mock

tests/unit/test_client_timeouts.py

@@ -116,7 +116,7 @@ def assert_timeout(expected_timeout: int, request: Request) -> Response:
     (RequestQueueClient, 'delete_request', request_queue._SMALL_TIMEOUT, {'request_id': 123}),
     (RequestQueueClient, 'prolong_request_lock', request_queue._MEDIUM_TIMEOUT, {'request_id': 123, 'lock_secs': 1}),
     (RequestQueueClient, 'delete_request_lock', request_queue._SMALL_TIMEOUT, {'request_id': 123}),
-    (RequestQueueClient, 'batch_add_requests', request_queue._MEDIUM_TIMEOUT, {'requests': [{'uniqueKey': '123'}]}),
+    (RequestQueueClient, 'batch_add_requests', request_queue._MEDIUM_TIMEOUT, {'requests': [{}]}),
     (RequestQueueClient, 'batch_delete_requests', request_queue._SMALL_TIMEOUT, {'requests': [{}]}),
     (RequestQueueClient, 'list_requests', request_queue._MEDIUM_TIMEOUT, {}),
 ]