From f63b4fa9814b3c105e3ca44b66a06fc133f37b65 Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Fri, 14 Aug 2026 12:03:30 +0000 Subject: [PATCH] apply throttling behind a tandem and with retry_on_blocked disabled --- .../_abstract_http/_abstract_http_crawler.py | 12 +-- src/crawlee/crawlers/_basic/_basic_crawler.py | 83 ++++++++++++------- .../_playwright/_playwright_crawler.py | 12 +-- .../_request_manager_tandem.py | 5 ++ .../crawlers/_basic/test_basic_crawler.py | 64 +++++++++++--- .../unit/crawlers/_http/test_http_crawler.py | 36 ++++++++ 6 files changed, 156 insertions(+), 56 deletions(-) diff --git a/src/crawlee/crawlers/_abstract_http/_abstract_http_crawler.py b/src/crawlee/crawlers/_abstract_http/_abstract_http_crawler.py index fb85d71d4d..deef38f27e 100644 --- a/src/crawlee/crawlers/_abstract_http/_abstract_http_crawler.py +++ b/src/crawlee/crawlers/_abstract_http/_abstract_http_crawler.py @@ -303,13 +303,13 @@ async def _handle_status_code_response( The original crawling context if no errors are detected. """ status_code = context.http_response.status_code + self._record_rate_limit_status_code( + status_code, + request_url=context.request.url, + retry_after_header=context.http_response.headers.get('retry-after'), + ) if self._retry_on_blocked: - self._raise_for_session_blocked_status_code( - context.session, - status_code, - request_url=context.request.url, - retry_after_header=context.http_response.headers.get('retry-after'), - ) + self._raise_for_session_blocked_status_code(context.session, status_code) self._raise_for_error_status_code(status_code) yield context diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index 96ff205350..8a3188ba78 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -65,7 +65,7 @@ ) from crawlee.events._types import Event, EventCrawlerStatusData from crawlee.http_clients import ImpitHttpClient -from crawlee.request_loaders import ThrottlingRequestManager +from crawlee.request_loaders import RequestManagerTandem, ThrottlingRequestManager from crawlee.router import Router from crawlee.sessions import SessionPool from crawlee.statistics import Statistics, StatisticsState @@ -705,7 +705,7 @@ async def run( self._running = True - if self._respect_robots_txt_file and not isinstance(self._request_manager, ThrottlingRequestManager): + if self._respect_robots_txt_file and self._get_throttling_manager() is None: self._logger.warning( 'The `respect_robots_txt_file` option is enabled, but the crawler is not using ' '`ThrottlingRequestManager`. Crawl-delay directives from robots.txt will not be enforced. To enable ' @@ -1619,51 +1619,69 @@ def _raise_for_error_status_code(self, status_code: int) -> None: if is_status_code_server_error(status_code) and not is_ignored_status: raise HttpStatusCodeError('Error status code returned', status_code) - def _raise_for_session_blocked_status_code( + def _get_throttling_manager(self) -> ThrottlingRequestManager | None: + """Return the crawler's `ThrottlingRequestManager`, unwrapping any `RequestManagerTandem` around it.""" + manager = self._request_manager + while isinstance(manager, RequestManagerTandem): + manager = manager.request_manager + + return manager if isinstance(manager, ThrottlingRequestManager) else None + + def _record_rate_limit_status_code( self, - session: Session | None, status_code: int, *, request_url: str, retry_after_header: str | None = None, ) -> None: - """Raise an exception if the given status code indicates the session is blocked. + """Record a 429 Too Many Requests response so the domain gets a per-domain backoff. - If the status code is 429 (Too Many Requests), the domain is recorded as rate-limited in the - `ThrottlingRequestManager` for per-domain backoff. + Other status codes are ignored. Rate limiting is independent of `retry_on_blocked`, so this runs for every + response, not only for those checked against the session's blocking rules. Args: - session: The session used for the request. If `None`, no check is performed. status_code: The HTTP status code to check. request_url: The request URL, used for per-domain rate limit tracking. retry_after_header: The value of the `Retry-After` response header, if present. - - Raises: - SessionError: If the status code indicates the session is blocked. """ - if status_code == HTTPStatus.TOO_MANY_REQUESTS: - if isinstance(self._request_manager, ThrottlingRequestManager): - retry_after = parse_retry_after_header(retry_after_header) - if not self._request_manager.record_domain_delay(request_url, retry_after=retry_after): - domain = (URL(request_url).host or '').lower() - if domain: - self._logger_once.log( - f'Received an HTTP 429 (Too Many Requests) response from domain "{domain}", but it is ' - f'not in the `ThrottlingRequestManager.domains` list. Per-domain backoff will not be ' - f'applied for this domain. Add it to `domains=` to enable throttling.', - key=f'unconfigured_throttle_domain:{domain}', - level=logging.WARNING, - ) - else: + if status_code != HTTPStatus.TOO_MANY_REQUESTS: + return + + throttling_manager = self._get_throttling_manager() + + if throttling_manager is None: + self._logger_once.log( + 'Received an HTTP 429 (Too Many Requests) response, but the crawler is not using ' + '`ThrottlingRequestManager`. Per-domain backoff and `Retry-After` headers will not be honored. ' + 'To enable per-domain rate limiting, configure the crawler to use `ThrottlingRequestManager` ' + 'as the request manager.', + key='no_throttling_manager_on_429', + level=logging.WARNING, + ) + return + + retry_after = parse_retry_after_header(retry_after_header) + if not throttling_manager.record_domain_delay(request_url, retry_after=retry_after): + domain = (URL(request_url).host or '').lower() + if domain: self._logger_once.log( - 'Received an HTTP 429 (Too Many Requests) response, but the crawler is not using ' - '`ThrottlingRequestManager`. Per-domain backoff and `Retry-After` headers will not be honored. ' - 'To enable per-domain rate limiting, configure the crawler to use `ThrottlingRequestManager` ' - 'as the request manager.', - key='no_throttling_manager_on_429', + f'Received an HTTP 429 (Too Many Requests) response from domain "{domain}", but it is ' + f'not in the `ThrottlingRequestManager.domains` list. Per-domain backoff will not be ' + f'applied for this domain. Add it to `domains=` to enable throttling.', + key=f'unconfigured_throttle_domain:{domain}', level=logging.WARNING, ) + def _raise_for_session_blocked_status_code(self, session: Session | None, status_code: int) -> None: + """Raise an exception if the given status code indicates the session is blocked. + + Args: + session: The session used for the request. If `None`, no check is performed. + status_code: The HTTP status code to check. + + Raises: + SessionError: If the status code indicates the session is blocked. + """ if session is not None and session.is_blocked_status_code( status_code=status_code, ignore_http_error_status_codes=self._ignore_http_error_status_codes, @@ -1697,10 +1715,11 @@ async def _is_allowed_based_on_robots_txt_file(self, url: str) -> bool: if not robots_txt_file: return True - if isinstance(self._request_manager, ThrottlingRequestManager): + throttling_manager = self._get_throttling_manager() + if throttling_manager is not None: crawl_delay = robots_txt_file.get_crawl_delay() if crawl_delay is not None: - self._request_manager.set_crawl_delay(url, crawl_delay) + throttling_manager.set_crawl_delay(url, crawl_delay) return robots_txt_file.is_allowed(url) diff --git a/src/crawlee/crawlers/_playwright/_playwright_crawler.py b/src/crawlee/crawlers/_playwright/_playwright_crawler.py index 1235b3675c..12bdd7646e 100644 --- a/src/crawlee/crawlers/_playwright/_playwright_crawler.py +++ b/src/crawlee/crawlers/_playwright/_playwright_crawler.py @@ -534,13 +534,13 @@ async def _handle_status_code_response(self, context: TPostNavContext) -> AsyncG The original crawling context if no errors are detected. """ status_code = context.response.status + self._record_rate_limit_status_code( + status_code, + request_url=context.request.url, + retry_after_header=context.response.headers.get('retry-after'), + ) if self._retry_on_blocked: - self._raise_for_session_blocked_status_code( - context.session, - status_code, - request_url=context.request.url, - retry_after_header=context.response.headers.get('retry-after'), - ) + self._raise_for_session_blocked_status_code(context.session, status_code) self._raise_for_error_status_code(status_code) yield context diff --git a/src/crawlee/request_loaders/_request_manager_tandem.py b/src/crawlee/request_loaders/_request_manager_tandem.py index a605c25d25..48c6610f5d 100644 --- a/src/crawlee/request_loaders/_request_manager_tandem.py +++ b/src/crawlee/request_loaders/_request_manager_tandem.py @@ -32,6 +32,11 @@ def __init__(self, request_loader: RequestLoader, request_manager: RequestManage self._read_only_loader = request_loader self._read_write_manager = request_manager + @property + def request_manager(self) -> RequestManager: + """The wrapped manager that stores the requests, both its own and those handed over by the loader.""" + return self._read_write_manager + @override async def get_handled_count(self) -> int: return await self._read_write_manager.get_handled_count() diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index 56ba257e86..3950667d9d 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -28,7 +28,7 @@ from crawlee.crawlers import BasicCrawler from crawlee.errors import RequestCollisionError, SessionError, UserDefinedErrorHandlerError from crawlee.events import Event, EventCrawlerStatusData, LocalEventManager -from crawlee.request_loaders import RequestList, RequestManagerTandem, ThrottlingRequestManager +from crawlee.request_loaders import RequestList, RequestManager, RequestManagerTandem, ThrottlingRequestManager from crawlee.sessions import Session, SessionPool from crawlee.statistics import FinalStatistics, StatisticsState from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient @@ -2464,8 +2464,8 @@ async def test_warn_no_throttling_manager_once_on_429(caplog: pytest.LogCaptureF """A 429 from a crawler without ThrottlingRequestManager logs a recommendation, only once per instance.""" crawler = BasicCrawler(configure_logging=False) with caplog.at_level(logging.WARNING, logger='crawlee'): - crawler._raise_for_session_blocked_status_code(session=None, status_code=429, request_url='https://a.test/') - crawler._raise_for_session_blocked_status_code(session=None, status_code=429, request_url='https://b.test/') + crawler._record_rate_limit_status_code(429, request_url='https://a.test/') + crawler._record_rate_limit_status_code(429, request_url='https://b.test/') matching = [ r for r in caplog.records if 'ThrottlingRequestManager' in r.getMessage() and 'HTTP 429' in r.getMessage() @@ -2484,15 +2484,9 @@ async def test_warn_unconfigured_throttle_domain_once_per_domain(caplog: pytest. crawler = BasicCrawler(configure_logging=False, request_manager=throttler) with caplog.at_level(logging.WARNING, logger='crawlee'): - crawler._raise_for_session_blocked_status_code( - session=None, status_code=429, request_url='https://A.example.com/page1' - ) - crawler._raise_for_session_blocked_status_code( - session=None, status_code=429, request_url='https://a.example.com/page2' - ) - crawler._raise_for_session_blocked_status_code( - session=None, status_code=429, request_url='https://other.example.com/page1' - ) + crawler._record_rate_limit_status_code(429, request_url='https://A.example.com/page1') + crawler._record_rate_limit_status_code(429, request_url='https://a.example.com/page2') + crawler._record_rate_limit_status_code(429, request_url='https://other.example.com/page1') matching = [ r @@ -2502,3 +2496,49 @@ async def test_warn_unconfigured_throttle_domain_once_per_domain(caplog: pytest. assert len(matching) == 2 assert any('a.example.com' in r.getMessage() for r in matching) assert any('other.example.com' in r.getMessage() for r in matching) + + +@pytest.mark.parametrize( + 'tandem_depth', + [ + pytest.param(1, id='single_tandem'), + pytest.param(2, id='nested_tandems'), + ], +) +async def test_records_429_through_tandem(tandem_depth: int, caplog: pytest.LogCaptureFixture) -> None: + """A throttler reached through a tandem still gets the 429, and the crawler stops claiming there is no throttler.""" + inner = await RequestQueue.open(alias=f'tandem-429-{tandem_depth}', storage_client=MemoryStorageClient()) + throttler = ThrottlingRequestManager( + inner, + domains=['throttled.test'], + request_manager_opener=RequestQueue.open, + ) + + manager: RequestManager = throttler + for depth in range(tandem_depth): + manager = await RequestList([f'https://loader{depth}.test/']).to_tandem(manager) + + crawler = BasicCrawler(configure_logging=False, request_manager=manager) + + with caplog.at_level(logging.WARNING, logger='crawlee'): + crawler._record_rate_limit_status_code(429, request_url='https://throttled.test/page1') + + assert throttler._is_domain_throttled('throttled.test') + assert not [r for r in caplog.records if 'not using' in r.getMessage()] + + +async def test_no_crawl_delay_warning_for_tandem_wrapped_throttler(caplog: pytest.LogCaptureFixture) -> None: + """The robots.txt crawl-delay warning must not fire when the throttler is wrapped in a tandem.""" + inner = await RequestQueue.open(alias='tandem-crawl-delay', storage_client=MemoryStorageClient()) + throttler = ThrottlingRequestManager( + inner, + domains=['throttled.test'], + request_manager_opener=RequestQueue.open, + ) + tandem = await RequestList([]).to_tandem(throttler) + crawler = BasicCrawler(configure_logging=False, request_manager=tandem, respect_robots_txt_file=True) + + with caplog.at_level(logging.WARNING, logger='crawlee'): + await crawler.run() + + assert not [r for r in caplog.records if 'Crawl-delay directives' in r.getMessage()] diff --git a/tests/unit/crawlers/_http/test_http_crawler.py b/tests/unit/crawlers/_http/test_http_crawler.py index 7d6fbfda0b..5e86d313ab 100644 --- a/tests/unit/crawlers/_http/test_http_crawler.py +++ b/tests/unit/crawlers/_http/test_http_crawler.py @@ -9,6 +9,7 @@ from crawlee import ConcurrencySettings, Request, RequestState from crawlee.crawlers import HttpCrawler +from crawlee.request_loaders import ThrottlingRequestManager from crawlee.sessions import SessionPool from crawlee.statistics import Statistics from crawlee.storages import RequestQueue @@ -686,3 +687,38 @@ async def failed_request_handler(context: BasicCrawlingContext, _error: Exceptio } await queue.drop() + + +@pytest.mark.parametrize( + 'retry_on_blocked', + [ + pytest.param(True, id='retry_on_blocked'), + pytest.param(False, id='no_retry_on_blocked'), + ], +) +async def test_records_429_regardless_of_retry_on_blocked( + mock_request_handler: AsyncMock, + server_url: URL, + *, + retry_on_blocked: bool, +) -> None: + """Rate limiting is a separate concern from session blocking, so a 429 must be recorded either way.""" + domain = server_url.host or '' + inner = await RequestQueue.open(alias=f'throttle-429-{retry_on_blocked}') + throttler = ThrottlingRequestManager( + inner, + domains=[domain], + request_manager_opener=RequestQueue.open, + ) + crawler = HttpCrawler( + request_handler=mock_request_handler, + request_manager=throttler, + retry_on_blocked=retry_on_blocked, + max_request_retries=0, + # Without this, a 429 retires the session and the retries walk the backoff up to `max_delay`. + use_session_pool=False, + ) + + await crawler.run([str(server_url / 'status/429')]) + + assert throttler._is_domain_throttled(domain)