From ea105da9dc7dc217c0c9c9590db20f286d8891d6 Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Mon, 2 Nov 2026 22:51:40 +0000 Subject: [PATCH] make `inner` optional in `ThrottlingRequestManager` --- docs/03_concepts/19_request_throttling.mdx | 2 +- .../request_throttling/throttling_example.py | 8 +- docs/06_upgrading/upgrading_to_v2.md | 4 + .../_throttling_request_manager.py | 116 ++++++++++----- .../crawlers/_basic/test_basic_crawler.py | 6 +- tests/unit/test_throttling_request_manager.py | 133 ++++++++++++++++++ 6 files changed, 224 insertions(+), 45 deletions(-) diff --git a/docs/03_concepts/19_request_throttling.mdx b/docs/03_concepts/19_request_throttling.mdx index 76cadf37b1..8ab738b270 100644 --- a/docs/03_concepts/19_request_throttling.mdx +++ b/docs/03_concepts/19_request_throttling.mdx @@ -13,7 +13,7 @@ When crawling websites that enforce rate limits (HTTP 429) or specify `crawl-del ## Overview -The `ThrottlingRequestManager` wraps a `RequestManager` (typically a `RequestQueue`) and manages per-domain throttling. You specify which domains to throttle at initialization, and the manager automatically: +The `ThrottlingRequestManager` wraps a `RequestManager` (typically a `RequestQueue`) and manages per-domain throttling. If you don't pass one to wrap, it opens the default request queue on first use. You specify which domains to throttle at initialization, and the manager automatically: - **Routes requests** for listed domains into dedicated sub-managers at insertion time. - **Enforces delays** from HTTP 429 responses (exponential backoff) and `robots.txt` crawl-delay directives. diff --git a/docs/03_concepts/code_examples/request_throttling/throttling_example.py b/docs/03_concepts/code_examples/request_throttling/throttling_example.py index 40e278bb8d..ec80fe9272 100644 --- a/docs/03_concepts/code_examples/request_throttling/throttling_example.py +++ b/docs/03_concepts/code_examples/request_throttling/throttling_example.py @@ -6,13 +6,9 @@ async def main() -> None: - # Open the default request queue. - queue = await RequestQueue.open() - - # Wrap it with ThrottlingRequestManager for specific domains. The throttler uses the - # same storage backend as the underlying queue. + # Throttle specific domains. Without a queue to wrap, the throttler opens the default + # request queue on first use. throttler = ThrottlingRequestManager( - queue, domains=['api.example.com', 'slow-site.org'], request_manager_opener=RequestQueue.open, ) diff --git a/docs/06_upgrading/upgrading_to_v2.md b/docs/06_upgrading/upgrading_to_v2.md index c20d72a3a8..dc8bd52e41 100644 --- a/docs/06_upgrading/upgrading_to_v2.md +++ b/docs/06_upgrading/upgrading_to_v2.md @@ -473,6 +473,10 @@ Entering `Statistics` or `SessionPool` again keeps their in-memory state. In v1, `persistence_enabled` of `Statistics` no longer accepts `'explicit_only'`. Pass `True` or `False`. +## `ThrottlingRequestManager.inner` is optional + +The `inner` argument of `ThrottlingRequestManager` is now optional, the same as in Crawlee JS. Without it, `request_manager_opener` opens the default store on first use, such as the default request queue for `RequestQueue.open`. Until then, the `inner` property returns `None`, so its type is now `TRequestManager | None`. If you pass `inner`, the property returns it as before. + ## Logs changed If you configure loggers by name or match log messages, update them for the following changes. diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py index 928d2448e4..eabde23b05 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -67,7 +67,8 @@ class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]): The class is generic over the wrapped manager type. The first asynchronous operation opens one sub-manager per configured domain through `request_manager_opener`, so all of them share the subclass and backing store of `inner`; `record_pacing_signal` never opens anything. The opener must accept `alias`, `storage_backend`, and `configuration` - keyword arguments (as `RequestQueue.open` does) and return the same concrete subclass as `inner`. + keyword arguments (as `RequestQueue.open` does) and return the same concrete subclass as `inner`. Without `inner`, + the opener also opens the default store, such as the default request queue, on first use. Requests a previous run left in a persistent store become visible again at open. The default `purge_on_start=True` empties them; `purge_on_start=False` resumes them. Named stores are exempt from that purge and aliased ones are not, @@ -80,9 +81,7 @@ class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]): from crawlee.request_loaders import ThrottlingRequestManager from crawlee.storages import RequestQueue - queue = await RequestQueue.open() throttler = ThrottlingRequestManager( - inner=queue, domains=['api.example.com', 'slow-site.org'], request_manager_opener=RequestQueue.open, ) @@ -92,7 +91,7 @@ class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]): def __init__( self, - inner: TRequestManager, + inner: TRequestManager | None = None, *, domains: Sequence[str], request_manager_opener: _RequestManagerOpener[TRequestManager], @@ -105,7 +104,7 @@ def __init__( Args: inner: The underlying request manager to wrap (typically a `RequestQueue`). Requests for non-throttled - domains are stored here. + domains are stored here. If omitted, `request_manager_opener` opens the default one on first use. domains: Domains to throttle, each given as a bare hostname such as `api.example.com`, or as any URL on the domain, of which only the hostname is used. Blank entries are ignored, so a list built by splitting a string needs no pruning. Only requests matching these domains will be routed to @@ -114,9 +113,10 @@ def __init__( supported — list each subdomain explicitly if needed. request_manager_opener: Async callable used to open one sub-manager per configured domain on first use. Must accept `alias`, `storage_backend`, and `configuration` keyword arguments and return the same - concrete subclass as `inner` (e.g. `RequestQueue.open` when `inner` is a `RequestQueue`). - service_locator: Service locator for creating sub-managers. If not provided, defaults to the global service - locator, ensuring consistency with the crawler's storage backend. + concrete subclass as `inner` (e.g. `RequestQueue.open` when `inner` is a `RequestQueue`). Without + `inner`, it's also called without `alias` to open the default one. + service_locator: Service locator for creating sub-managers and the default `inner`. If not provided, + defaults to the global service locator, ensuring consistency with the crawler's storage backend. base_delay: Initial delay after the first 429 response from a domain. Must be positive. max_delay: Maximum delay between requests to a rate-limited domain. Must be positive. max_domain_stall: How long a domain may rate-limit every request before the crawler raises @@ -137,7 +137,8 @@ def __init__( if max_domain_stall <= timedelta(0): raise ValueError(f'max_domain_stall must be positive, got {max_domain_stall}.') - self._inner: TRequestManager = inner + self._inner = inner + self._inner_lock = asyncio.Lock() self._service_locator = service_locator if service_locator is not None else global_service_locator self._base_delay = base_delay self._max_delay = max_delay @@ -158,30 +159,35 @@ def __init__( self._migrated_from_inner = 0 """Number of requests moved from `inner` into a sub-manager. Both stores count such a request, so the handled and total counts subtract it.""" + self._expected_request_processing_time: timedelta | None = None + """The longest `set_expected_request_processing_time` value, kept for a default `inner` opened later.""" @property - def inner(self) -> TRequestManager: - """The wrapped request manager that stores requests for non-throttled domains.""" + def inner(self) -> TRequestManager | None: + """The wrapped request manager that stores requests for non-throttled domains. + + `None` until the default one is opened on first use, if no `inner` was passed. + """ return self._inner @override async def drop(self) -> None: await self._ensure_sub_managers() - await asyncio.gather(self._inner.drop(), *(sm.drop() for sm in self._sub_managers.values())) + await asyncio.gather(*(manager.drop() for manager in self._opened_managers())) self._sub_managers.clear() self._sub_managers_ready = False self._in_flight_from_inner.clear() @override async def purge(self) -> None: - """Empty the inner manager and all sub-managers, and reset transient per-domain throttle state. + """Empty the inner manager, if it's open, and all sub-managers, and reset transient per-domain throttle state. The configured domain list and any robots.txt-derived `crawl_delay` are preserved. Only the dynamic backoff state (consecutive 429 counter, the throttle clocks and the stall clock) is cleared. Sub-managers stay open; they're just emptied. """ await self._ensure_sub_managers() - await asyncio.gather(self._inner.purge(), *(sm.purge() for sm in self._sub_managers.values())) + await asyncio.gather(*(manager.purge() for manager in self._opened_managers())) self._in_flight_from_inner.clear() self._migrated_from_inner = 0 for state in self._domain_states.values(): @@ -189,16 +195,20 @@ async def purge(self) -> None: @override async def set_expected_request_processing_time(self, processing_time: timedelta) -> None: + # A request queue keeps only the longest value, so the one kept for a later `inner` follows suit. Otherwise a + # shorter value set last would reach it alone and cut the locks of a consumer that holds requests longer. + kept = self._expected_request_processing_time + self._expected_request_processing_time = processing_time if kept is None else max(kept, processing_time) await self._ensure_sub_managers() await asyncio.gather( - self._inner.set_expected_request_processing_time(processing_time), - *(sm.set_expected_request_processing_time(processing_time) for sm in self._sub_managers.values()), + *(manager.set_expected_request_processing_time(processing_time) for manager in self._opened_managers()) ) @override async def extend_request_processing_time(self, request: Request, delta: timedelta) -> bool: await self._ensure_sub_managers() - return await self._fetch_owner(request).extend_request_processing_time(request, delta) + manager = await self._fetch_owner(request) + return await manager.extend_request_processing_time(request, delta) @override async def add_request(self, request: str | Request, *, forefront: bool = False) -> ProcessedRequest | None: @@ -215,7 +225,8 @@ async def add_request(self, request: str | Request, *, forefront: bool = False) if domain in self._domain_states: return await self._sub_managers[domain].add_request(request, forefront=forefront) - return await self._inner.add_request(request, forefront=forefront) + inner = await self._get_inner() + return await inner.add_request(request, forefront=forefront) @override async def add_requests( @@ -244,7 +255,8 @@ async def add_requests( inner_requests.append(request) if inner_requests: - await self._inner.add_requests( + inner = await self._get_inner() + await inner.add_requests( inner_requests, forefront=forefront, batch_size=batch_size, @@ -290,7 +302,7 @@ async def fetch_next_request(self) -> Request | None: @override async def reclaim_request(self, request: Request, *, forefront: bool = False) -> ProcessedRequest | None: await self._ensure_sub_managers() - manager = self._fetch_owner(request) + manager = await self._fetch_owner(request) result = await manager.reclaim_request(request, forefront=forefront) self._clear_fetch_owner(request) return result @@ -303,7 +315,7 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | state = self._get_domain_state(request.url) if state is not None: state.rate_limited_since = None - manager = self._fetch_owner(request) + manager = await self._fetch_owner(request) result = await manager.mark_request_as_handled(request) self._clear_fetch_owner(request) return result @@ -311,24 +323,27 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | @override async def get_handled_count(self) -> int: await self._ensure_sub_managers() + inner = await self._get_inner() counts = await asyncio.gather( - self._inner.get_handled_count(), *(sm.get_handled_count() for sm in self._sub_managers.values()) + inner.get_handled_count(), *(sm.get_handled_count() for sm in self._sub_managers.values()) ) return sum(counts) - self._migrated_from_inner @override async def get_total_count(self) -> int: await self._ensure_sub_managers() + inner = await self._get_inner() counts = await asyncio.gather( - self._inner.get_total_count(), *(sm.get_total_count() for sm in self._sub_managers.values()) + inner.get_total_count(), *(sm.get_total_count() for sm in self._sub_managers.values()) ) return sum(counts) - self._migrated_from_inner @override async def get_pending_count(self) -> int: await self._ensure_sub_managers() + inner = await self._get_inner() counts = await asyncio.gather( - self._inner.get_pending_count(), *(sm.get_pending_count() for sm in self._sub_managers.values()) + inner.get_pending_count(), *(sm.get_pending_count() for sm in self._sub_managers.values()) ) # `inner` counts a migrated request as handled, so it is pending only in its sub-manager and needs no fix-up. return sum(counts) @@ -343,8 +358,9 @@ async def check_readiness(self) -> RequestSourceStatus: requests in progress or still being added. """ await self._ensure_sub_managers() + inner = await self._get_inner() now = datetime.now(UTC) - dispatchable: list[RequestManager] = [self._inner] + dispatchable: list[RequestManager] = [inner] cooling: list[str] = [] candidates: list[str] = [] @@ -599,10 +615,37 @@ async def _ensure_sub_managers(self) -> None: self._sub_managers_ready = True + async def _get_inner(self) -> TRequestManager: + """Return `inner`, opening the default one through `request_manager_opener` on first use.""" + if self._inner is not None: + return self._inner + + async with self._inner_lock: + inner = self._inner + if inner is None: + inner = await self._request_manager_opener( + storage_backend=self._service_locator.get_storage_backend(), + configuration=self._service_locator.get_configuration(), + ) + # Set before the hint is applied, so a hint that arrives meanwhile reaches it directly. + self._inner = inner + if self._expected_request_processing_time is not None: + await inner.set_expected_request_processing_time(self._expected_request_processing_time) + + return inner + + def _opened_managers(self) -> list[TRequestManager]: + """Return the opened sub-managers and `inner`, without opening the default `inner` just for this.""" + managers = list(self._sub_managers.values()) + if self._inner is not None: + managers.append(self._inner) + return managers + async def _fetch_from_inner(self) -> Request | None: """Fetch the next request from `inner`, moving each one whose domain is in a cooldown into its sub-manager.""" + inner = await self._get_inner() for _ in range(_MAX_INNER_MIGRATIONS): - request = await self._inner.fetch_next_request() + request = await inner.fetch_next_request() if request is None: return None @@ -610,11 +653,11 @@ async def _fetch_from_inner(self) -> Request | None: if state is None or datetime.now(UTC) >= state.throttled_until: return request - await self._migrate_to_sub_manager(request) + await self._migrate_to_sub_manager(request, inner) return None - async def _migrate_to_sub_manager(self, request: Request) -> None: + async def _migrate_to_sub_manager(self, request: Request, inner: TRequestManager) -> None: """Move a request fetched from `inner` into its domain's sub-manager.""" sub_manager = self._sub_managers[self._extract_domain(request.url)] try: @@ -622,15 +665,15 @@ async def _migrate_to_sub_manager(self, request: Request) -> None: processed = await sub_manager.add_request(request.model_copy(deep=True)) # Includes cancellation, which would otherwise leave the request in progress in `inner`. except BaseException: - await self._inner.reclaim_request(request, forefront=True) + await inner.reclaim_request(request, forefront=True) raise # A storage that refuses the request returns `None` instead of raising. if processed is None: - await self._inner.reclaim_request(request, forefront=True) + await inner.reclaim_request(request, forefront=True) raise RuntimeError(f'The storage refused to move request {request.url} into its domain queue.') - await self._inner.mark_request_as_handled(request) + await inner.mark_request_as_handled(request) self._migrated_from_inner += 1 def _fetchable_domains(self) -> list[str]: @@ -649,14 +692,15 @@ def _mark_domain_dispatched(self, domain: str) -> None: if state is not None: state.apply_crawl_delay(datetime.now(UTC)) - def _fetch_owner(self, request: Request) -> TRequestManager: + async def _fetch_owner(self, request: Request) -> TRequestManager: """Return the manager the request must be given back to, leaving its in-flight record in place. `_clear_fetch_owner` drops the record only once the completion is accepted, so a retry resolves the same way. """ - if (request.unique_key, request.url) in self._in_flight_from_inner: - return self._inner - return self._sub_managers.get(self._extract_domain(request.url), self._inner) + sub_manager = self._sub_managers.get(self._extract_domain(request.url)) + if sub_manager is None or (request.unique_key, request.url) in self._in_flight_from_inner: + return await self._get_inner() + return sub_manager def _clear_fetch_owner(self, request: Request) -> None: """Drop the in-flight record of a request whose completion was accepted.""" @@ -668,7 +712,7 @@ class _RequestManagerOpener(Protocol[TRequestManager_co]): Matches the keyword-only signature shared by storage `open` classmethods such as `RequestQueue.open`. `ThrottlingRequestManager` invokes the opener at sub-manager creation time, so every sub-manager shares the same - backing type as `inner`. + backing type as `inner`. Without an `alias`, it opens the default `inner`. """ async def __call__( diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index 093a7215c2..f8ff759706 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -3448,7 +3448,9 @@ async def test_stall_behind_tandem_raises() -> None: async def test_inner_held_429_is_deferred_and_stalls() -> None: """A rate-limited request held by `inner` moves into its domain's sub-manager and can stall the crawl.""" throttler = await open_throttler() - await throttler.inner.add_request(THROTTLED_URL) + inner = throttler.inner + assert inner is not None + await inner.add_request(THROTTLED_URL) crawler = BasicCrawler(request_manager=throttler) calls = 0 @@ -3463,7 +3465,7 @@ async def handler(_context: BasicCrawlingContext) -> None: with pytest.raises(PersistentRateLimitError): await crawler.run() - assert (await throttler.inner.check_readiness()).status == 'finished' + assert (await inner.check_readiness()).status == 'finished' assert (await throttler.check_readiness()).status != 'finished' diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py index c01398f4e1..cccd47a41b 100644 --- a/tests/unit/test_throttling_request_manager.py +++ b/tests/unit/test_throttling_request_manager.py @@ -1080,6 +1080,138 @@ async def test_sub_managers_opened_once( assert opener.await_count == len(domains) +async def test_default_inner_opened_on_first_use( + service_locator: ServiceLocator, + memory_storage_backend: MemoryStorageBackend, +) -> None: + """Test that a manager without `inner` opens the default store through the opener and routes requests there.""" + opener = AsyncMock(side_effect=RequestQueue.open) + manager = ThrottlingRequestManager( + domains=TEST_DOMAINS, + request_manager_opener=opener, + service_locator=service_locator, + ) + assert manager.inner is None + + await manager.add_request(f'https://{NON_THROTTLED_DOMAIN}/page1') + + opener.assert_any_await(storage_backend=memory_storage_backend, configuration=service_locator.get_configuration()) + assert manager.inner is not None + assert (await manager.inner.check_readiness()).status == 'ready' + + +async def test_default_inner_opened_once(service_locator: ServiceLocator) -> None: + """Test that concurrent first calls open the default `inner` exactly once.""" + + # Suspends before opening, so the other calls arrive while the default `inner` is still being opened. + async def open_after_yield(**kwargs: Any) -> RequestQueue: + await asyncio.sleep(0) + return await RequestQueue.open(**kwargs) + + opener = AsyncMock(side_effect=open_after_yield) + manager = ThrottlingRequestManager( + domains=TEST_DOMAINS, + request_manager_opener=opener, + service_locator=service_locator, + ) + + await asyncio.gather(manager.get_total_count(), manager.get_handled_count(), manager.get_pending_count()) + + assert opener.await_count == len(TEST_DOMAINS) + 1 + + +@pytest.mark.parametrize( + 'method', + [ + pytest.param('purge', id='purge'), + pytest.param('drop', id='drop'), + ], +) +async def test_default_inner_not_opened_to_clear(service_locator: ServiceLocator, method: str) -> None: + """Test that purging or dropping doesn't open the default `inner` just to clear it.""" + manager = ThrottlingRequestManager( + domains=TEST_DOMAINS, + request_manager_opener=RequestQueue.open, + service_locator=service_locator, + ) + + await getattr(manager, method)() + + assert manager.inner is None + + +async def test_processing_time_reaches_default_inner(service_locator: ServiceLocator) -> None: + """Test that an expected processing time set before the default `inner` opens is applied to it.""" + manager = ThrottlingRequestManager( + domains=TEST_DOMAINS, + request_manager_opener=RequestQueue.open, + service_locator=service_locator, + ) + + with patch.object(RequestQueue, 'set_expected_request_processing_time', autospec=True) as set_processing_time: + await manager.set_expected_request_processing_time(timedelta(minutes=10)) + assert manager.inner is None + await manager.add_request(f'https://{NON_THROTTLED_DOMAIN}/page1') + + set_processing_time.assert_any_await(manager.inner, timedelta(minutes=10)) + + +async def test_default_inner_gets_longest_time(service_locator: ServiceLocator) -> None: + """Test that a default `inner` opened late gets the longest processing time set before, as an open one keeps.""" + manager = ThrottlingRequestManager( + domains=TEST_DOMAINS, + request_manager_opener=RequestQueue.open, + service_locator=service_locator, + ) + + with patch.object(RequestQueue, 'set_expected_request_processing_time', autospec=True) as set_processing_time: + await manager.set_expected_request_processing_time(timedelta(minutes=10)) + await manager.set_expected_request_processing_time(timedelta(minutes=5)) + await manager.add_request(f'https://{NON_THROTTLED_DOMAIN}/page1') + + awaits = set_processing_time.await_args_list + inner_times = [awaited.args[1] for awaited in awaits if awaited.args[0] is manager.inner] + assert inner_times == [timedelta(minutes=10)] + + +async def test_check_readiness_opens_default_inner(service_locator: ServiceLocator) -> None: + """Test that `check_readiness` opens the default `inner`, so requests already in it aren't missed.""" + default_queue = await RequestQueue.open( + storage_backend=service_locator.get_storage_backend(), + configuration=service_locator.get_configuration(), + ) + await default_queue.add_request(f'https://{NON_THROTTLED_DOMAIN}/left-behind') + manager = ThrottlingRequestManager( + domains=TEST_DOMAINS, + request_manager_opener=RequestQueue.open, + service_locator=service_locator, + ) + + assert (await manager.check_readiness()).status == 'ready' + assert manager.inner is default_queue + + +async def test_default_inner_open_retried_after_failure(service_locator: ServiceLocator) -> None: + """Test that a failed open of the default `inner` is retried on the next use.""" + default_queue = await RequestQueue.open( + storage_backend=service_locator.get_storage_backend(), + configuration=service_locator.get_configuration(), + ) + # Without configured domains, the opener is called only for the default `inner`. + manager = ThrottlingRequestManager( + domains=[], + request_manager_opener=AsyncMock(side_effect=[RuntimeError('storage unavailable'), default_queue]), + service_locator=service_locator, + ) + + with pytest.raises(RuntimeError, match='storage unavailable'): + await manager.check_readiness() + assert manager.inner is None + + assert (await manager.check_readiness()).status == 'finished' + assert manager.inner is default_queue + + async def test_failed_open_keeps_successful_sub_managers( inner_queue: RequestQueue, service_locator: ServiceLocator, @@ -1203,6 +1335,7 @@ async def test_reclaim_routes_to_sub_manager_after_restart(fs_service_locator: S await restarted.reclaim_request(request) assert (await restarted._sub_managers[THROTTLED_DOMAIN].check_readiness()).status == 'ready' + assert restarted.inner is not None assert (await restarted.inner.check_readiness()).status == 'finished'