Skip to content
Merged
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
12 changes: 12 additions & 0 deletions docs/upgrading/upgrading_to_v2.md
Original file line number Diff line number Diff line change
Expand Up @@ -167,3 +167,15 @@ 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.

## Only the first crawler uses the default request queue

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.

Should there be some warning about multiprocessing cases breaking some assumptions and thus not behaving the same way out of the box?

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.

I don't think so. Our documentation doesn't contain any recommendations regarding multiprocessing.


Only the first crawler created in a process opens the default request queue, the same as Crawlee JS. Every later crawler without a `request_manager` opens its own queue under the alias `__default_{id}__`. A crawler without an `id` uses its creation index there, such as `__default_1__`. In v1, all these crawlers shared the default queue, so a URL one crawler handled was skipped by the others.

To reach a crawler's queue, call `crawler.get_request_manager()` instead of `RequestQueue.open()`. If your crawlers relied on the shared queue, open it yourself and pass it to each of them:

```python
queue = await RequestQueue.open()
crawler_1 = BasicCrawler(request_manager=queue)
crawler_2 = BasicCrawler(request_manager=queue)
```
14 changes: 9 additions & 5 deletions src/crawlee/crawlers/_basic/_basic_crawler.py
Original file line number Diff line number Diff line change
Expand Up @@ -224,8 +224,9 @@ class _BasicCrawlerOptions(TypedDict):
Returning `None` suppresses the status message."""

id: NotRequired[str]
"""A unique identifier for the crawler instance. It isolates the `use_state` state and the default statistics from
other crawlers. Without it, the ID follows the order in which crawlers are created."""
"""Identifier of the crawler. Crawlers with the same ID share the `use_state` state and the default statistics, and
every crawler but the first uses it in the alias of its default request queue. Without it, the ID follows the order
in which crawlers are created."""


class _BasicCrawlerOptionsGeneric(TypedDict, Generic[TCrawlingContext, TStatisticsState]):
Expand Down Expand Up @@ -371,8 +372,9 @@ def __init__(
status_message_logging_interval: Interval for logging the crawler status messages.
status_message_callback: Allows overriding the default status message. The default status message is
provided in the parameters. Returning `None` suppresses the status message.
id: A unique identifier for the crawler instance. It isolates the `use_state` state and the default
statistics from other crawlers. Without it, the ID follows the order in which crawlers are created.
id: Identifier of the crawler. Crawlers with the same ID share the `use_state` state and the default
statistics, and every crawler but the first uses it in the alias of its default request queue. Without
it, the ID follows the order in which crawlers are created.
_context_pipeline: Enables extending the request lifecycle and modifying the crawling context.
Intended for use by subclasses rather than direct instantiation of `BasicCrawler`.
_additional_context_managers: Additional context managers used throughout the crawler lifecycle.
Expand All @@ -381,6 +383,7 @@ def __init__(
Intended for use by subclasses rather than direct instantiation of `BasicCrawler`.
"""
self._id = str(BasicCrawler.__next_id) if id is None else id
self._default_rq_alias = None if BasicCrawler.__next_id == 0 else f'__default_{self._id}__'

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.

self._default_rq_alias = None

Just double-checking since now is a good time for whatever breaking changes, are we sure that the default RQ should not have any alias?

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.

I'd keep it. Many users run a single crawler on the default queue, and this way nothing changes for them. Also, on the Apify platform, the default queue is the run's default storage, so with an alias the first crawler's requests would go to a separate unnamed queue instead of the one the Console shows for the run.

BasicCrawler.__next_id += 1
self._consumer = ConcurrencyConsumer(id=self._id)

Expand Down Expand Up @@ -644,9 +647,10 @@ async def _get_proxy_info(self, request: Request, session: Session | None) -> Pr
)

async def get_request_manager(self) -> RequestManager:
"""Return the configured request manager. If none is configured, open and return the default request queue."""
"""Return the configured request manager. If none is configured, open and return the crawler's own queue."""
if not self._request_manager:
self._request_manager = await RequestQueue.open(
alias=self._default_rq_alias,
storage_client=self._service_locator.get_storage_client(),
configuration=self._service_locator.get_configuration(),
)
Expand Down
33 changes: 33 additions & 0 deletions tests/unit/crawlers/_basic/test_basic_crawler.py
Original file line number Diff line number Diff line change
Expand Up @@ -1270,6 +1270,39 @@ def test_default_statistics_key_differs_from_later_statistics() -> None:
assert statistics._state._persist_state_key != crawler.statistics._state._persist_state_key


async def test_only_first_crawler_uses_default_request_queue() -> None:
"""Only the first crawler opens the default RQ, the later ones open their own under an alias from their ID."""
crawler_1 = BasicCrawler(id='first')
crawler_2 = BasicCrawler()
crawler_3 = BasicCrawler(id='custom')

assert await crawler_1.get_request_manager() is await RequestQueue.open()
assert await crawler_2.get_request_manager() is await RequestQueue.open(alias='__default_1__')
assert await crawler_3.get_request_manager() is await RequestQueue.open(alias='__default_custom__')
Comment thread
Mantisus marked this conversation as resolved.


async def test_crawlers_do_not_share_requests() -> None:
"""Two crawlers without a request manager each crawl the same URL, since they don't share a queue."""
crawler_1 = BasicCrawler(request_handler=AsyncMock())
crawler_2 = BasicCrawler(request_handler=AsyncMock())

stats_1 = await crawler_1.run(['https://a.placeholder.com'])
stats_2 = await crawler_2.run(['https://a.placeholder.com'])

assert stats_1.requests_finished == 1
assert stats_2.requests_finished == 1


async def test_id_invalid_as_request_queue_alias() -> None:
"""A later crawler whose ID can't form an RQ alias fails to open its RQ, unless it gets a request manager."""
BasicCrawler()
with pytest.raises(ValueError, match='Invalid storage alias'):
await BasicCrawler(id='shop/products').get_request_manager()
request_manager = Mock()
crawler = BasicCrawler(id='shop/products', request_manager=request_manager)
assert await crawler.get_request_manager() is request_manager


async def test_context_handlers_use_state(key_value_store: KeyValueStore) -> None:
state_in_handler_one: dict[str, JsonSerializable] = {}
state_in_handler_two: dict[str, JsonSerializable] = {}
Expand Down
Loading