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'