refactor(request-loaders)!: Replace is_empty and is_finished with check_readiness

docs/examples/code_examples/run_parallel_crawlers.py

@@ -79,7 +79,7 @@ async def handle_parsel(context: ParselCrawlingContext) -> None:
         await parsel_crawler.run(['https://crawlee.dev/blog'])
 
         # Wait for the Playwright crawler to finish processing all requests
-        while not await playwright_rq.is_empty():
+        while (await playwright_rq.check_readiness()).status != 'finished':
             playwright_crawler.log.info('Waiting for Playwright crawler to finish...')
             await asyncio.sleep(5)
 

docs/guides/request_loaders.mdx

@@ -62,8 +62,7 @@ class RequestLoader {
     + total_count
     + fetch_next_request()
     + mark_request_as_handled()
-    + is_empty()
-    + is_finished()
+    + check_readiness()
     + to_tandem()
 }
 

docs/upgrading/upgrading_to_v2.md

@@ -95,3 +95,33 @@ The `Request.forefront` property is removed. Proxy tiers were the only code that
 The storage clients now write their JSON data with camelCase keys, the same as Crawlee JS. This covers the metadata, request and record metadata files of `FileSystemStorageClient`, the request and record metadata JSON stored by `RedisStorageClient`, and the request JSON stored by `SqlStorageClient`. For example, `item_count` in a dataset's `__metadata__.json` becomes `itemCount`. The storage metadata documents of `RedisStorageClient` keep their snake_case keys.
 
 Data written by v1 still loads, so an existing storage directory, Redis instance or database keeps working. Any tooling that reads these files directly needs to handle the camelCase keys.
+
+## `check_readiness` replaces `is_empty` and `is_finished`
+
+`RequestLoader` and `RequestManager` no longer have `is_empty` and `is_finished`. A single `check_readiness` method replaces them and returns one of four statuses:
+
+- `ReadyStatus`: the next `fetch_next_request` call is expected to return a request.
+- `WaitingStatus`: nothing can be fetched right now, but the source isn't done. Requests are in progress, are being added in the background, or are held back until `ready_at`.
+- `StalledStatus`: the source holds requests it can't make progress on, as explained in `reason`. `ThrottlingRequestManager` reports it, and `RequestManagerTandem` passes it on from the manager it wraps.
+- `FinishedStatus`: every request has been handled.
+
+Each of these classes also has a `status` attribute set to `'ready'`, `'waiting'`, `'stalled'` or `'finished'`, so a check doesn't need an import.
+
+The change covers `RequestQueue`, `RequestList`, `SitemapRequestLoader`, `RequestManagerTandem` and `ThrottlingRequestManager`. `ThrottlingRequestManager.get_stall_reason` is removed too, since a stall is now reported as `StalledStatus`. Storage clients keep `is_empty` and `is_finished`.
+
+```python
+# v1
+has_requests = not await rq.is_empty()
+done = await rq.is_finished()
+
+# v2
+readiness = await rq.check_readiness()
+has_requests = readiness.status == 'ready'
+done = readiness.status == 'finished'
+```
+
+If you have a custom request loader or manager, replace `is_empty` and `is_finished` with `check_readiness`. Return `ReadyStatus` as soon as a request can be fetched, before checking anything else, since the crawler calls the method often. Never return `FinishedStatus` while requests are still pending or in progress. A loader doesn't return `StalledStatus`.
+
+A `None` from `fetch_next_request` no longer means the loader is finished. It only means nothing is pending right now. `RequestManagerTandem` now moves a request from its loader only when the loader is ready, so a `SitemapRequestLoader` that's still loading sitemaps no longer holds up the requests already in the manager.
+
+The crawler now reads a stall from the status, so it also sees a `ThrottlingRequestManager` wrapped in a `RequestManagerTandem`. Such a crawl now ends with `PersistentRateLimitError` once a domain stalls, as it already did with a `ThrottlingRequestManager` passed directly. Before, the crawler didn't detect the stall. The error it logs before raising `PersistentRateLimitError` now reads "Giving up the crawl because the request manager can't make progress." instead of "Giving up the crawl because a domain keeps rate-limiting every request."

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -1453,21 +1453,17 @@ async def __is_finished_function(self) -> bool:
         if self._keep_alive:
             return False
 
+        request_manager = await self.get_request_manager()
+        readiness = await request_manager.check_readiness()
+
         # Check for a stall only when the pool is idle. An in-flight request can still enqueue new work or end the
         # stall.
-        if (
-            isinstance(self._request_manager, ThrottlingRequestManager)
-            and self._autoscaled_pool.current_concurrency == 0
-            and (reason := await self._request_manager.get_stall_reason()) is not None
-        ):
-            error = PersistentRateLimitError(f'Giving up: {reason}')
-            self._logger.error(
-                'Giving up the crawl because a domain keeps rate-limiting every request.', exc_info=error
-            )
+        if readiness.status == 'stalled' and self._autoscaled_pool.current_concurrency == 0:
+            error = PersistentRateLimitError(f'Giving up: {readiness.reason}')
+            self._logger.error("Giving up the crawl because the request manager can't make progress.", exc_info=error)
             raise error
 
-        request_manager = await self.get_request_manager()
-        return await request_manager.is_finished()
+        return readiness.status == 'finished'
 
     async def __is_task_ready_function(self) -> bool:
         self._stop_if_max_requests_count_exceeded()
@@ -1479,7 +1475,7 @@ async def __is_task_ready_function(self) -> bool:
             return False
 
         request_manager = await self.get_request_manager()
-        return not await request_manager.is_empty()
+        return (await request_manager.check_readiness()).status == 'ready'
 
     async def __run_task_function(self) -> None:
         request_manager = await self.get_request_manager()

src/crawlee/request_loaders/__init__.py

@@ -4,12 +4,26 @@
 from ._request_manager_tandem import RequestManagerTandem
 from ._sitemap_request_loader import SitemapRequestLoader
 from ._throttling_request_manager import ThrottlingRequestManager
+from ._types import (
+    FinishedStatus,
+    ReadyStatus,
+    RequestLoaderStatus,
+    RequestSourceStatus,
+    StalledStatus,
+    WaitingStatus,
+)
 
 __all__ = [
+    'FinishedStatus',
+    'ReadyStatus',
     'RequestList',
     'RequestLoader',
+    'RequestLoaderStatus',
     'RequestManager',
     'RequestManagerTandem',
+    'RequestSourceStatus',
     'SitemapRequestLoader',
+    'StalledStatus',
     'ThrottlingRequestManager',
+    'WaitingStatus',
 ]

src/crawlee/request_loaders/_request_list.py

@@ -4,7 +4,7 @@
 import contextlib
 from collections.abc import AsyncGenerator, AsyncIterable, AsyncIterator, Iterable
 from logging import getLogger
-from typing import Annotated
+from typing import TYPE_CHECKING, Annotated
 
 from pydantic import BaseModel, ConfigDict, Field, ValidationError
 from pydantic.alias_generators import to_camel
@@ -14,6 +14,11 @@
 from crawlee._utils.docs import docs_group
 from crawlee.request_loaders._request_loader import RequestLoader
 
+from ._types import FinishedStatus, ReadyStatus, WaitingStatus
+
+if TYPE_CHECKING:
+    from ._types import RequestLoaderStatus
+
 logger = getLogger(__name__)
 
 
@@ -155,14 +160,13 @@ async def get_total_count(self) -> int:
         return self._assumed_total_count
 
     @override
-    async def is_empty(self) -> bool:
+    async def check_readiness(self) -> RequestLoaderStatus:
         await self._ensure_next_request()
-        return self._next[0] is None
+        if self._next[0] is not None:
+            return ReadyStatus()
 
-    @override
-    async def is_finished(self) -> bool:
         state = await self._get_state()
-        return len(state.in_progress) == 0 and await self.is_empty()
+        return WaitingStatus() if state.in_progress else FinishedStatus()
 
     @override
     async def fetch_next_request(self) -> Request | None:

src/crawlee/request_loaders/_request_loader.py

@@ -12,6 +12,8 @@
     from crawlee.request_loaders import RequestManager, RequestManagerTandem
     from crawlee.storage_clients.models import ProcessedRequest
 
+    from ._types import RequestSourceStatus
+
 
 @docs_group('Request loaders')
 class RequestLoader(ABC):
@@ -34,19 +36,18 @@ async def get_total_count(self) -> int:
         """Get an offline approximation of the total number of requests in the loader (i.e. pending + handled)."""
 
     @abstractmethod
-    async def is_empty(self) -> bool:
-        """Return True if there are no more requests in the loader (there might still be unfinished requests)."""
+    async def check_readiness(self) -> RequestSourceStatus:
+        """Report whether the loader has a request to hand over, is waiting for one, or is done.
 
-    @abstractmethod
-    async def is_finished(self) -> bool:
-        """Return True if all requests have been handled."""
+        The crawler's task loop calls this often, so answer `ReadyStatus` before evaluating anything else.
+        `FinishedStatus` may arrive late, but never early.
+        """
 
     @abstractmethod
     async def fetch_next_request(self) -> Request | None:
-        """Return the next request to be processed, or `None` if there are no more pending requests.
+        """Return the next request to be processed, or `None` if no request is pending right now.
 
-        The method should return `None` if and only if `is_finished` would return `True`. In other cases, the method
-        should wait until a request appears.
+        `None` doesn't mean the loader is finished. Use `check_readiness` to tell a finished loader from a waiting one.
         """
 
     @abstractmethod

src/crawlee/request_loaders/_request_manager_tandem.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from datetime import timedelta
 from logging import getLogger
 from typing import TYPE_CHECKING
@@ -9,13 +10,17 @@
 from crawlee._utils.docs import docs_group
 from crawlee.request_loaders import RequestManager
 
+from ._utils import join_request_source_statuses
+
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
     from crawlee import Request
     from crawlee.request_loaders import RequestLoader
     from crawlee.storage_clients.models import ProcessedRequest
 
+    from ._types import RequestSourceStatus
+
 
 logger = getLogger(__name__)
 
@@ -41,12 +46,11 @@ async def get_total_count(self) -> int:
         return (await self._read_only_loader.get_total_count()) + (await self._read_write_manager.get_total_count())
 
     @override
-    async def is_empty(self) -> bool:
-        return (await self._read_only_loader.is_empty()) and (await self._read_write_manager.is_empty())
-
-    @override
-    async def is_finished(self) -> bool:
-        return (await self._read_only_loader.is_finished()) and (await self._read_write_manager.is_finished())
+    async def check_readiness(self) -> RequestSourceStatus:
+        loader_status, manager_status = await asyncio.gather(
+            self._read_only_loader.check_readiness(), self._read_write_manager.check_readiness()
+        )
+        return join_request_source_statuses(loader_status, manager_status)
 
     @override
     async def add_request(self, request: str | Request, *, forefront: bool = False) -> ProcessedRequest | None:
@@ -74,13 +78,22 @@ async def add_requests(
 
     @override
     async def fetch_next_request(self) -> Request | None:
-        if await self._read_only_loader.is_finished():
-            return await self._read_write_manager.fetch_next_request()
+        # Only the loader's own readiness decides the transfer, so a loader that's still loading doesn't block the
+        # manager's requests.
+        if (await self._read_only_loader.check_readiness()).status == 'ready':
+            transferred = await self._transfer_next_request()
+            # The request was dropped, so skip the manager this round. The next call picks up where this one left off.
+            if not transferred:
+                return None
+
+        return await self._read_write_manager.fetch_next_request()
 
+    async def _transfer_next_request(self) -> bool:
+        """Move the loader's next request into the manager, returning `False` if it had to be dropped."""
         request = await self._read_only_loader.fetch_next_request()
 
         if not request:
-            return await self._read_write_manager.fetch_next_request()
+            return True
 
         try:
             await self._read_write_manager.add_request(request, forefront=True)
@@ -90,13 +103,13 @@ async def fetch_next_request(self) -> Request | None:
                 extra={'url': request.url, 'unique_key': request.unique_key},
             )
 
-            return None
+            return False
         finally:
             # Mark it as processed so that the `request` doesn't get stuck in the `in_progress` status
             # in `RequestLoader`
             await self._read_only_loader.mark_request_as_handled(request)
 
-        return await self._read_write_manager.fetch_next_request()
+        return True
 
     @override
     async def reclaim_request(self, request: Request, *, forefront: bool = False) -> None:

src/crawlee/request_loaders/_sitemap_request_loader.py

@@ -26,6 +26,8 @@
 from crawlee._utils.urls import filter_url
 from crawlee.request_loaders._request_loader import RequestLoader
 
+from ._types import FinishedStatus, ReadyStatus, WaitingStatus
+
 if TYPE_CHECKING:
     import re
     from collections.abc import Callable, Sequence
@@ -37,6 +39,8 @@
     from crawlee.proxy_configuration import ProxyInfo
     from crawlee.storage_clients.models import ProcessedRequest
 
+    from ._types import RequestLoaderStatus
+
 
 logger = getLogger(__name__)
 
@@ -206,23 +210,26 @@ async def get_handled_count(self) -> int:
         return state.handled_count
 
     @override
-    async def is_empty(self) -> bool:
-        """Check if there are no more URLs to process."""
+    async def check_readiness(self) -> RequestLoaderStatus:
         state = await self._get_state()
-        return not state.url_queue
+        if state.url_queue:
+            return ReadyStatus()
 
-    @override
-    async def is_finished(self) -> bool:
-        """Check if all URLs have been processed."""
-        state = await self._get_state()
-        return not state.url_queue and len(state.in_progress) == 0 and self._loading_task.done()
+        # More URLs may arrive while the sitemaps are still being loaded.
+        if not self._loading_task.done():
+            return WaitingStatus()
+
+        return WaitingStatus() if state.in_progress else FinishedStatus()
 
     @override
     async def fetch_next_request(self) -> Request | None:
         """Fetch the next request to process."""
-        while not (await self.is_finished()):
+        while True:
             state = await self._get_state()
             if not state.url_queue:
+                # More URLs may arrive while the sitemaps are still being loaded.
+                if self._loading_task.done():
+                    return None
                 await asyncio.sleep(0.1)
                 continue
 
@@ -250,8 +257,6 @@ async def fetch_next_request(self) -> Request | None:
 
             return request
 
-        return None
-
     @override
     async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None:
         """Mark a request as successfully handled."""
@@ -449,6 +454,6 @@ async def _load_sitemaps(self) -> None:
             state.completed = True
 
         except Exception:
-            # Not re-raised: nothing retrieves the task's exception (`is_finished` only checks doneness), so it
+            # Not re-raised: nothing retrieves the task's exception (`check_readiness` only checks doneness), so it
             # would only resurface as an "exception was never retrieved" asyncio error when the task is collected.
             logger.exception('Error loading sitemaps')