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

@@ -167,3 +167,35 @@ current concurrency 3, desired 5, overloaded: cpu 58% of the time (limit 40%)
 To read the concurrency values in code, use `crawler.concurrency_system`.
 
 The autoscaling loggers moved together with the package. `crawlee._autoscaling.autoscaled_pool`, `crawlee._autoscaling.snapshotter` and `crawlee._autoscaling.system_status` are now `crawlee.autoscaling._autoscaled_pool`, `crawlee.autoscaling._snapshotter` and `crawlee.autoscaling._system_status`. The state log, along with the messages about scaling and free capacity, now comes from the new `crawlee.autoscaling._concurrency_system` logger. To configure all of them at once, use the `crawlee.autoscaling` logger.
+
+## `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."
+
+The two DEBUG messages that `RequestQueue.is_finished` logged, "Background add requests tasks are still in progress." and "The request queue is finished.", are removed together with the method.

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -1499,22 +1499,21 @@ 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. The pool's own requests are what counts, not those of another crawler sharing the budget.
         if (
-            isinstance(self._request_manager, ThrottlingRequestManager)
+            readiness.status == 'stalled'
             and self._autoscaled_pool is not None
             and self._autoscaled_pool.own_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
-            )
+            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()
@@ -1526,7 +1525,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, task_id: str) -> None:  # noqa: ARG002
         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

@@ -9,13 +9,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 +45,13 @@ 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())
+    async def check_readiness(self) -> RequestSourceStatus:
+        loader_readiness = await self._read_only_loader.check_readiness()
+        if loader_readiness.status == 'ready':
+            return loader_readiness
 
-    @override
-    async def is_finished(self) -> bool:
-        return (await self._read_only_loader.is_finished()) and (await self._read_write_manager.is_finished())
+        manager_readiness = await self._read_write_manager.check_readiness()
+        return join_request_source_statuses(loader_readiness, manager_readiness)
 
     @override
     async def add_request(self, request: str | Request, *, forefront: bool = False) -> ProcessedRequest | None:
@@ -74,7 +79,9 @@ async def add_requests(
 
     @override
     async def fetch_next_request(self) -> Request | None:
-        if await self._read_only_loader.is_finished():
+        # 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':
             return await self._read_write_manager.fetch_next_request()
 
         request = await self._read_only_loader.fetch_next_request()

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')