refactor(request-loaders)!: Replace `is_empty` and `is_finished` with `check_readiness` by Mantisus · Pull Request #2334 · apify/crawlee-python · GitHub
Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/examples/code_examples/run_parallel_crawlers.py
3 changes: 1 addition & 2 deletions docs/guides/request_loaders.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -62,8 +62,7 @@ class RequestLoader {
+ total_count
+ fetch_next_request()
+ mark_request_as_handled()
+ is_empty()
+ is_finished()
+ check_readiness()
+ to_tandem()
}

Expand Down
32 changes: 32 additions & 0 deletions docs/upgrading/upgrading_to_v2.md
Original file line number Diff line number Diff line change
Expand Up @@ -179,3 +179,35 @@ queue = await RequestQueue.open()
crawler_1 = BasicCrawler(request_manager=queue)
crawler_2 = BasicCrawler(request_manager=queue)
```

## `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.
17 changes: 8 additions & 9 deletions src/crawlee/crawlers/_basic/_basic_crawler.py
Original file line number Diff line number Diff line change
Expand Up @@ -1503,22 +1503,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()
Expand All @@ -1530,7 +1529,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'

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note: The pool calls __is_finished_function and then __is_task_ready_function on every pass, and both now run check_readiness(). On a queue that is empty but not finished, each call makes two backend calls (is_empty() + is_finished()), so a pass costs 4 calls where it used to cost 2. That covers the end of a crawl, waiting on in-progress requests, and throttler cooldowns. On the platform these are API calls, and ThrottlingRequestManager repeats them for every domain queue. Could the status from __is_finished_function be reused here, or is that too tied to the call order in the pool?

✍️ Drafted by Claude Code

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It will be resolved in #2332


async def __run_task_function(self, task_id: str) -> None: # noqa: ARG002
request_manager = await self.get_request_manager()
Expand Down
14 changes: 14 additions & 0 deletions src/crawlee/request_loaders/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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',
]
16 changes: 10 additions & 6 deletions src/crawlee/request_loaders/_request_list.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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__)


Expand Down Expand Up @@ -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:
Expand Down
17 changes: 9 additions & 8 deletions src/crawlee/request_loaders/_request_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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
Expand Down
19 changes: 13 additions & 6 deletions src/crawlee/request_loaders/_request_manager_tandem.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)

Expand All @@ -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:
Expand Down Expand Up @@ -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()
Expand Down
29 changes: 17 additions & 12 deletions src/crawlee/request_loaders/_sitemap_request_loader.py
Loading
Loading