From 91efd1fb87e0fc0d001da72f63c92b9a371b30ce Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Fri, 6 Nov 2026 15:47:09 +0000 Subject: [PATCH 1/3] add to and raise for a request it isn't processing --- docs/06_upgrading/upgrading_to_v2.md | 4 +++ .../_sitemap_request_loader.py | 21 ++++++++++--- .../test_sitemap_request_loader.py | 31 +++++++++++++++++++ 3 files changed, 51 insertions(+), 5 deletions(-) diff --git a/docs/06_upgrading/upgrading_to_v2.md b/docs/06_upgrading/upgrading_to_v2.md index c20d72a3a8..05b544cd92 100644 --- a/docs/06_upgrading/upgrading_to_v2.md +++ b/docs/06_upgrading/upgrading_to_v2.md @@ -454,6 +454,10 @@ class MyHttpClient(HttpClient): `RequestManagerTandem.get_total_count` no longer counts a request twice once it moves from the loader to the manager. In v1, it added up the totals of both, so the crawler's status message showed too many pages in total. +## `SitemapRequestLoader.mark_request_as_handled` raises for an unknown request + +`SitemapRequestLoader.mark_request_as_handled` now raises a `ValueError` for a request that the loader isn't processing, such as one it never returned or one that's already handled. Crawlee JS raises in the same case. In v1, such a call did nothing. + ## Resetting statistics and the session pool `Statistics.reset()` is now synchronous and resets only the in-memory statistics, the same as Crawlee JS. In v1, it also cleared the persisted record. To clear the record too, call the new `reset_store()` once the statistics are exited: diff --git a/src/crawlee/request_loaders/_sitemap_request_loader.py b/src/crawlee/request_loaders/_sitemap_request_loader.py index f0a6af0647..8d0345e617 100644 --- a/src/crawlee/request_loaders/_sitemap_request_loader.py +++ b/src/crawlee/request_loaders/_sitemap_request_loader.py @@ -281,12 +281,18 @@ async def fetch_next_request(self) -> Request | None: @override async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None: - """Mark a request as successfully handled.""" + """Mark a request as successfully handled. + + Raises: + ValueError: If the request isn't being processed. + """ state = await self._get_state() - if request.url in state.in_progress: - state.in_progress.remove(request.url) - del state.request_data[request.url] - state.handled_count += 1 + if request.url not in state.in_progress: + raise ValueError(f'The request is not being processed (url: {request.url}).') + + state.in_progress.remove(request.url) + del state.request_data[request.url] + state.handled_count += 1 return None async def start(self) -> None: @@ -302,6 +308,11 @@ async def abort_loading(self) -> None: with suppress(asyncio.CancelledError): await self._loading_task + async def persist_state(self) -> None: + """Persist the loader's state to the key-value store. Does nothing without a `persist_state_key`.""" + await self._get_state() + await self._state.persist_state() + async def close(self) -> None: """Close the request loader.""" self._closed = True diff --git a/tests/unit/request_loaders/test_sitemap_request_loader.py b/tests/unit/request_loaders/test_sitemap_request_loader.py index 48596f8a12..24f829a8eb 100644 --- a/tests/unit/request_loaders/test_sitemap_request_loader.py +++ b/tests/unit/request_loaders/test_sitemap_request_loader.py @@ -217,6 +217,37 @@ def to_api_url(request_options: RequestOptions) -> RequestOptions: assert restored_request.user_data['page'] == 2 +async def test_persist_state_writes_record(key_value_store: KeyValueStore) -> None: + """`persist_state()` writes the state with the in-progress request to the key-value store.""" + sitemap_url = f'{DEFAULT_URL}sitemap.xml' + client, _ = make_status_stream_client({sitemap_url: [(200, get_basic_sitemap().encode())]}) + loader = SitemapRequestLoader([sitemap_url], http_client=client, persist_state_key='sitemap-state') + request = await loader.fetch_next_request() + assert request is not None + + await loader.persist_state() + record = await key_value_store.get_value('sitemap-state') + await loader.close() + + assert record['inProgress'] == [request.url] + assert list(record['requestData']) == [request.url] + + +async def test_mark_handled_twice_raises() -> None: + """Marking a request that's already handled raises and leaves the handled count as is.""" + sitemap_url = f'{DEFAULT_URL}sitemap.xml' + client, _ = make_status_stream_client({sitemap_url: [(200, get_basic_sitemap().encode())]}) + loader = SitemapRequestLoader([sitemap_url], http_client=client) + request = await loader.fetch_next_request() + assert request is not None + await loader.mark_request_as_handled(request) + + with pytest.raises(ValueError, match='not being processed'): + await loader.mark_request_as_handled(request) + + assert await loader.get_handled_count() == 1 + + async def test_close_before_first_use_stops_listening(monkeypatch: pytest.MonkeyPatch) -> None: """A loader closed before its first use doesn't persist on PERSIST_STATE events once it's used.""" sitemap_url = f'{DEFAULT_URL}sitemap.xml' From 2f2d1044103831ae14e1b2db2006665462bd23db Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Mon, 9 Nov 2026 23:01:50 +0000 Subject: [PATCH 2/3] fix requests duplicates --- docs/06_upgrading/upgrading_to_v2.md | 2 +- .../_sitemap_request_loader.py | 13 ++++-- .../test_sitemap_request_loader.py | 41 +++++++++++++++++++ 3 files changed, 52 insertions(+), 4 deletions(-) diff --git a/docs/06_upgrading/upgrading_to_v2.md b/docs/06_upgrading/upgrading_to_v2.md index 05b544cd92..6b29b7b93e 100644 --- a/docs/06_upgrading/upgrading_to_v2.md +++ b/docs/06_upgrading/upgrading_to_v2.md @@ -454,7 +454,7 @@ class MyHttpClient(HttpClient): `RequestManagerTandem.get_total_count` no longer counts a request twice once it moves from the loader to the manager. In v1, it added up the totals of both, so the crawler's status message showed too many pages in total. -## `SitemapRequestLoader.mark_request_as_handled` raises for an unknown request +## `SitemapRequestLoader.mark_request_as_handled` raises for a request it isn't processing `SitemapRequestLoader.mark_request_as_handled` now raises a `ValueError` for a request that the loader isn't processing, such as one it never returned or one that's already handled. Crawlee JS raises in the same case. In v1, such a call did nothing. diff --git a/src/crawlee/request_loaders/_sitemap_request_loader.py b/src/crawlee/request_loaders/_sitemap_request_loader.py index 8d0345e617..4138e6db01 100644 --- a/src/crawlee/request_loaders/_sitemap_request_loader.py +++ b/src/crawlee/request_loaders/_sitemap_request_loader.py @@ -62,8 +62,9 @@ class SitemapRequestLoaderState(BaseModel): `processed_sitemap_urls`. If `pending_sitemap_urls` is empty, `completed` is set to `True`. When `fetch_next_request` is called, a URL is extracted from `url_queue`, and the request built from it is added - to `request_data` and its URL to `in_progress`. When `mark_request_as_handled` is called for the request, it is - removed from both and `handled_count` is incremented by 1. + to `request_data` and its URL to `in_progress`. A URL that's already in `in_progress`, because another sitemap + lists it too, is skipped and `total_count` is decremented by 1. When `mark_request_as_handled` is called for the + request, it is removed from both and `handled_count` is incremented by 1. During initial startup or restart after persistence, state validation occurs in `_get_state`. If both `pending_sitemap_urls` and `in_progress_sitemap_url` are empty and `completed` is False, this indicates a @@ -261,7 +262,8 @@ async def fetch_next_request(self) -> Request | None: self._queue_has_capacity.set() # A request restored as in progress is served again as it was. - if (restored_request := state.request_data.get(url)) is not None: + restored_request = state.request_data.get(url) + if restored_request is not None and url not in state.in_progress: state.in_progress.add(url) return restored_request @@ -274,6 +276,10 @@ async def fetch_next_request(self) -> Request | None: request_option = transform_request_option request = Request.from_url(**request_option) + # The URL was listed again while its request is still being processed. + if request.url in state.in_progress: + state.total_count -= 1 + continue state.in_progress.add(request.url) state.request_data[request.url] = request @@ -310,6 +316,7 @@ async def abort_loading(self) -> None: async def persist_state(self) -> None: """Persist the loader's state to the key-value store. Does nothing without a `persist_state_key`.""" + # Loads the persisted record first, so that a call before the first use writes the restored state. await self._get_state() await self._state.persist_state() diff --git a/tests/unit/request_loaders/test_sitemap_request_loader.py b/tests/unit/request_loaders/test_sitemap_request_loader.py index 24f829a8eb..7e8fd891b5 100644 --- a/tests/unit/request_loaders/test_sitemap_request_loader.py +++ b/tests/unit/request_loaders/test_sitemap_request_loader.py @@ -248,6 +248,47 @@ async def test_mark_handled_twice_raises() -> None: assert await loader.get_handled_count() == 1 +async def test_duplicate_url_skipped_while_in_progress() -> None: + """A URL listed again by another sitemap isn't served while its request is still in progress.""" + sitemap_body = f'{DEFAULT_URL}page'.encode() + sitemap_urls = [f'{DEFAULT_URL}sitemap-1.xml', f'{DEFAULT_URL}sitemap-2.xml'] + client, _ = make_status_stream_client({url: [(200, sitemap_body)] for url in sitemap_urls}) + loader = SitemapRequestLoader(sitemap_urls, http_client=client) + + request = await loader.fetch_next_request() + assert request is not None + duplicate = await loader.fetch_next_request() + await loader.mark_request_as_handled(request) + + assert duplicate is None + assert await loader.get_total_count() == 1 + assert await loader.get_handled_count() == 1 + + +async def test_restored_request_served_once_for_duplicate_url() -> None: + """A request restored as in progress isn't served again for the same URL listed by another sitemap.""" + sitemap_body = f'{DEFAULT_URL}page'.encode() + sitemap_urls = [f'{DEFAULT_URL}sitemap-1.xml', f'{DEFAULT_URL}sitemap-2.xml'] + client, _ = make_status_stream_client({url: [(200, sitemap_body)] for url in sitemap_urls}) + loader = SitemapRequestLoader(sitemap_urls, http_client=client, persist_state_key='sitemap-state') + request = await loader.fetch_next_request() + assert request is not None + # The second sitemap queues the URL again while its request is in progress. + assert await poll_until_condition(loader.get_pending_count, lambda count: count == 2) == 2 + await loader.close() + + restored_loader = SitemapRequestLoader(sitemap_urls, http_client=client, persist_state_key='sitemap-state') + restored_request = await restored_loader.fetch_next_request() + duplicate = await restored_loader.fetch_next_request() + total_count = await restored_loader.get_total_count() + await restored_loader.close() + + assert restored_request is not None + assert restored_request.url == request.url + assert duplicate is None + assert total_count == 1 + + async def test_close_before_first_use_stops_listening(monkeypatch: pytest.MonkeyPatch) -> None: """A loader closed before its first use doesn't persist on PERSIST_STATE events once it's used.""" sitemap_url = f'{DEFAULT_URL}sitemap.xml' From 8aee396796f8ead52e8f71c37e9f59a7471bcd96 Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Mon, 9 Nov 2026 23:13:52 +0000 Subject: [PATCH 3/3] clean --- .../request_loaders/_sitemap_request_loader.py | 6 ------ .../test_sitemap_request_loader.py | 16 ---------------- 2 files changed, 22 deletions(-) diff --git a/src/crawlee/request_loaders/_sitemap_request_loader.py b/src/crawlee/request_loaders/_sitemap_request_loader.py index 4138e6db01..5f80f28b3d 100644 --- a/src/crawlee/request_loaders/_sitemap_request_loader.py +++ b/src/crawlee/request_loaders/_sitemap_request_loader.py @@ -314,12 +314,6 @@ async def abort_loading(self) -> None: with suppress(asyncio.CancelledError): await self._loading_task - async def persist_state(self) -> None: - """Persist the loader's state to the key-value store. Does nothing without a `persist_state_key`.""" - # Loads the persisted record first, so that a call before the first use writes the restored state. - await self._get_state() - await self._state.persist_state() - async def close(self) -> None: """Close the request loader.""" self._closed = True diff --git a/tests/unit/request_loaders/test_sitemap_request_loader.py b/tests/unit/request_loaders/test_sitemap_request_loader.py index 7e8fd891b5..c72a315a18 100644 --- a/tests/unit/request_loaders/test_sitemap_request_loader.py +++ b/tests/unit/request_loaders/test_sitemap_request_loader.py @@ -217,22 +217,6 @@ def to_api_url(request_options: RequestOptions) -> RequestOptions: assert restored_request.user_data['page'] == 2 -async def test_persist_state_writes_record(key_value_store: KeyValueStore) -> None: - """`persist_state()` writes the state with the in-progress request to the key-value store.""" - sitemap_url = f'{DEFAULT_URL}sitemap.xml' - client, _ = make_status_stream_client({sitemap_url: [(200, get_basic_sitemap().encode())]}) - loader = SitemapRequestLoader([sitemap_url], http_client=client, persist_state_key='sitemap-state') - request = await loader.fetch_next_request() - assert request is not None - - await loader.persist_state() - record = await key_value_store.get_value('sitemap-state') - await loader.close() - - assert record['inProgress'] == [request.url] - assert list(record['requestData']) == [request.url] - - async def test_mark_handled_twice_raises() -> None: """Marking a request that's already handled raises and leaves the handled count as is.""" sitemap_url = f'{DEFAULT_URL}sitemap.xml'