Method _process_batch

src/crawlee/storages/_request_queue.py:347–396

Process a batch of requests with automatic retry mechanism. Returns: A response aggregating all requests processed across attempts plus any still unprocessed once the retries are exhausted.

(
        self,
        batch: Sequence[Request],
        *,
        base_retry_wait: timedelta,
        attempt: int = 1,
        forefront: bool = False,
    )

        return False

    async def _process_batch(
        self,
        batch: Sequence[Request],
        *,
        base_retry_wait: timedelta,
        attempt: int = 1,
        forefront: bool = False,
    ) -> AddRequestsResponse:
        """Process a batch of requests with automatic retry mechanism.

        Returns:
            A response aggregating all requests processed across attempts plus any still unprocessed once the
            retries are exhausted.
        """
        max_attempts = 5
        response = await self._client.add_batch_of_requests(batch, forefront=forefront)

        request_count = len(batch) - len(response.unprocessed_requests)
        if request_count:
            logger.debug(
                f'Added {request_count} requests to the queue. Processed requests: {response.processed_requests}'
            )

        if not response.unprocessed_requests:
            return response

        logger.debug(f'Following requests were not processed: {response.unprocessed_requests}.')
        if attempt > max_attempts:
            logger.warning(
                f'Following requests were not processed even after {max_attempts} attempts:\n'
                f'{response.unprocessed_requests}'
            )
            return response

        logger.debug('Retry to add requests.')
        unprocessed_requests_unique_keys = {request.unique_key for request in response.unprocessed_requests}
        retry_batch = [request for request in batch if request.unique_key in unprocessed_requests_unique_keys]
        await asyncio.sleep((base_retry_wait * attempt).total_seconds())
        retry_response = await self._process_batch(
            retry_batch,
            base_retry_wait=base_retry_wait,
            attempt=attempt + 1,
            forefront=forefront,
        )

        # Merge the retry outcome: processed requests accumulate, unprocessed is whatever the last attempt left.
        return AddRequestsResponse(
            processed_requests=[*response.processed_requests, *retry_response.processed_requests],
            unprocessed_requests=retry_response.unprocessed_requests,
        )