chore: Prevent request manager change after crawler restart

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -659,7 +659,10 @@ async def run(
             request_manager = await self.get_request_manager()
             if purge_request_queue and isinstance(request_manager, RequestQueue):
                 await request_manager.drop()
-                self._request_manager = await RequestQueue.open()
+                self._request_manager = await RequestQueue.open(
+                    storage_client=self._service_locator.get_storage_client(),
+                    configuration=self._service_locator.get_configuration(),
+                )
 
         if requests is not None:
             await self.add_requests(requests)

tests/unit/crawlers/_basic/test_basic_crawler.py

@@ -1620,3 +1620,26 @@ async def handler(context: BasicCrawlingContext) -> None:
             )
 
     await crawler.run(['https://start.placeholder.com'])
+
+
+async def test_crawler_purge_request_queue_uses_same_storage_client() -> None:
+    """Make sure that purge on start does not replace the storage client in the underlying storage manager"""
+
+    # Set some different storage_client globally and different for Crawlee.
+    service_locator.set_storage_client(FileSystemStorageClient())
+    unrelated_rq = await RequestQueue.open()
+    unrelated_request = Request.from_url('https://x.placeholder.com')
+    await unrelated_rq.add_request(unrelated_request)
+
+    crawler = BasicCrawler(storage_client=MemoryStorageClient())
+
+    @crawler.router.default_handler
+    async def handler(context: BasicCrawlingContext) -> None:
+        context.log.info(context.request.url)
+
+    for _ in (1, 2):
+        await crawler.run(requests=[Request.from_url('https://a.placeholder.com')], purge_request_queue=True)
+        assert crawler.statistics.state.requests_finished == 1
+
+    # Crawler should not fall back to the default storage after the purge
+    assert await unrelated_rq.fetch_next_request() == unrelated_request