From a9ed902e603ca2e02ac65ce29628ffce4f9d4921 Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Mon, 2 Nov 2026 00:11:10 +0000 Subject: [PATCH 1/3] extract per-run crawler state into `CrawlerRun` --- src/crawlee/crawlers/_basic/_basic_crawler.py | 113 +++++++++++------- src/crawlee/crawlers/_basic/_crawler_run.py | 82 +++++++++++++ .../crawlers/_basic/test_basic_crawler.py | 54 ++++++++- 3 files changed, 202 insertions(+), 47 deletions(-) create mode 100644 src/crawlee/crawlers/_basic/_crawler_run.py diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index 4b2dc37295..fec42e5c92 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -51,7 +51,7 @@ from crawlee._utils.urls import UNSUPPORTED_SCHEME_MESSAGE, convert_to_absolute_url, filter_url, is_url_absolute from crawlee._utils.wait import wait_for from crawlee._utils.web import is_status_code_client_error, is_status_code_server_error -from crawlee.autoscaling import AutoscaledPool, ConcurrencyConsumer, ConcurrencySystem +from crawlee.autoscaling import ConcurrencyConsumer, ConcurrencySystem from crawlee.errors import ( ContextPipelineInitializationError, ContextPipelineInterruptedError, @@ -74,6 +74,7 @@ from crawlee.storages import Dataset, KeyValueStore, RequestQueue from ._context_pipeline import ContextPipeline +from ._crawler_run import CrawlerRun from ._logging_utils import ( get_one_line_error_summary_if_possible, reduce_asyncio_timeout_error_to_relevant_traceback_parts, @@ -524,17 +525,15 @@ async def persist_state_factory() -> KeyValueStore: func=self._crawler_state_task, delay=status_message_logging_interval, ) - self._previous_crawler_state: TStatisticsState | None = None # State flags self._keep_alive = keep_alive - self._running = False self._has_finished_before = False - self._handled_request_this_run = False - self._failed = False - self._unexpected_stop = False + self._stop_before_run = False self._logger_once = LoggerOnce(self._logger) - self._run_logger_once = LoggerOnce(self._logger) + + # The last run started on this instance, which stays readable after it finishes. + self._last_run: CrawlerRun[TStatisticsState] | None = None @property def log(self) -> logging.Logger: @@ -570,6 +569,19 @@ def statistics(self) -> Statistics[TStatisticsState]: """Statistics about the current (or last) crawler run.""" return self._statistics + @property + def _run(self) -> CrawlerRun[TStatisticsState]: + """The last run started on this instance, for the code that only a run drives.""" + if self._last_run is None: + raise RuntimeError('No crawl has been started yet, only `run()` starts one.') + + return self._last_run + + @property + def _live_run(self) -> CrawlerRun[TStatisticsState] | None: + """The run in progress, if any.""" + return self._last_run if self._last_run is not None and self._last_run.is_live else None + def stop(self, reason: str = 'Stop was called externally.') -> None: """Set flag to stop crawler. @@ -579,7 +591,12 @@ def stop(self, reason: str = 'Stop was called externally.') -> None: reason: Reason for stopping that will be used in logs. """ self._logger.info(f'Crawler.stop() was called with following reason: {reason.rstrip(" .")}.') - self._unexpected_stop = True + + if self._live_run is not None: + self._live_run.stop() + else: + # `create_task(crawler.run())` doesn't start the coroutine yet, so the stop waits for the next run. + self._stop_before_run = True def _wrap_handler_with_error_context( self, handler: Callable[[TCrawlingContext | BasicCrawlingContext, Exception], Awaitable[T]] @@ -601,7 +618,7 @@ async def wrapped_handler(context: TCrawlingContext | BasicCrawlingContext, exce def _stop_if_max_requests_count_exceeded(self) -> None: """Call `stop` when the maximum number of requests to crawl has been reached.""" - if self._max_requests_per_crawl is None or self._unexpected_stop: + if self._max_requests_per_crawl is None or self._run.stop_requested: return if self._statistics.state.requests_total >= self._max_requests_per_crawl: @@ -732,13 +749,12 @@ async def run(self, requests: Sequence[str | Request] | None = None) -> FinalSta PersistentRateLimitError: If a domain throttled by a `ThrottlingRequestManager` has rate-limited every request for longer than the manager's `max_domain_stall` and no other requests are left. """ - if self._running: + if self._live_run is not None: raise RuntimeError( 'This crawler instance is already running, you can add more requests to it via `crawler.add_requests()`' ) - self._running = True - self._handled_request_this_run = False + self._last_run = self._create_run() try: if self._respect_robots_txt_file and not isinstance(self._request_manager, ThrottlingRequestManager): @@ -820,7 +836,7 @@ def sigint_handler() -> None: # A crawl that did nothing while the manager holds only handled requests is a mistake, whoever handled them. # Starting against handled requests is not, since that is what resuming a crawl looks like. The statistics # can't tell, as they don't count skipped requests and may be restored from a previous process. - if not self._handled_request_this_run: + if not self._run.handled_request: request_manager = await self.get_request_manager() # The diagnostic must never fail the run. try: @@ -844,37 +860,21 @@ def sigint_handler() -> None: return final_statistics finally: # A failed run must leave the instance usable, so that the caller can retry after handling the error. - self._running = False - - self._unexpected_stop = False - self._failed = False - self._run_logger_once = LoggerOnce(self._logger) - - async def _run_crawler(self) -> None: - local_event_manager = self._service_locator.get_event_manager() - global_event_manager = service_locator.get_event_manager() - if local_event_manager is global_event_manager: - local_event_manager = None # Avoid entering the same event manager context twice - - # The event managers are always entered. - contexts_to_enter: list[Any] = ( - [global_event_manager, local_event_manager] if local_event_manager else [global_event_manager] - ) + self._run.finish() - # A system supplied by the caller is only borrowed, and its lifecycle belongs to whoever created it. The one - # the crawler builds for this run is entered and exited below. + def _create_run(self) -> CrawlerRun[TStatisticsState]: if self._borrowed_concurrency_system is not None: concurrency_system: ConcurrencySystemInterface = self._borrowed_concurrency_system - owned_concurrency_system = None else: - owned_concurrency_system = ConcurrencySystem( + concurrency_system = ConcurrencySystem( self._concurrency_settings, configuration=self._service_locator.get_configuration(), ) - concurrency_system = owned_concurrency_system self._concurrency_system = concurrency_system - autoscaled_pool = AutoscaledPool( + + run = CrawlerRun[TStatisticsState]( + logger=self._logger, concurrency_system=concurrency_system, consumer=self._consumer, is_finished_function=self.__is_finished_function, @@ -882,6 +882,27 @@ async def _run_crawler(self) -> None: run_task_function=self.__run_task_function, ) + if self._stop_before_run: + run.stop() + self._stop_before_run = False + + return run + + async def _run_crawler(self) -> None: + local_event_manager = self._service_locator.get_event_manager() + global_event_manager = service_locator.get_event_manager() + if local_event_manager is global_event_manager: + local_event_manager = None # Avoid entering the same event manager context twice + + # The event managers are always entered. + contexts_to_enter: list[Any] = ( + [global_event_manager, local_event_manager] if local_event_manager else [global_event_manager] + ) + + # A system supplied by the caller is only borrowed, and its lifecycle belongs to whoever created it. The one + # the crawler builds for this run is entered and exited below. + owned_concurrency_system = self._concurrency_system if self._borrowed_concurrency_system is None else None + # Collect the context managers to be entered. Context managers that are already active are excluded, # as they were likely entered by the caller, who will also be responsible for exiting them. contexts_to_enter.extend( @@ -903,7 +924,7 @@ async def _run_crawler(self) -> None: for context in contexts_to_enter: await exit_stack.enter_async_context(context) - await autoscaled_pool.run() + await self._run.dispatch_requests() async def add_requests( self, @@ -1297,7 +1318,7 @@ async def _handle_request_retries( if self._abort_on_error: self._logger.error('Aborting crawler run due to error (abort_on_error=True)', exc_info=error) - self._failed = True + self._run.failed = True if self._should_retry_request(context, error): request.retry_count += 1 @@ -1500,11 +1521,11 @@ async def _commit_key_value_store_changes( async def __is_finished_function(self) -> bool: self._stop_if_max_requests_count_exceeded() - if self._unexpected_stop: + if self._run.stop_requested: self._logger.info('The crawler has finished all the remaining ongoing requests and will shut down now.') return True - if self._abort_on_error and self._failed: + if self._run.failed: return True if self._keep_alive: @@ -1524,8 +1545,8 @@ async def __is_finished_function(self) -> bool: async def __is_task_ready_function(self) -> bool: self._stop_if_max_requests_count_exceeded() - if self._unexpected_stop: - self._run_logger_once.log( + if self._run.stop_requested: + self._run.log_once( 'No new requests are allowed because the `stop()` method has been called. ' 'Ongoing requests will be allowed to complete.', key='shutting_down', @@ -1533,7 +1554,7 @@ async def __is_task_ready_function(self) -> bool: return False # The pool asks whether the run is finished only once nothing is in flight, so stop taking new requests here. - if self._abort_on_error and self._failed: + if self._run.failed: return False request_manager = await self.get_request_manager() @@ -1916,7 +1937,7 @@ async def _crawler_state_task(self) -> None: if ( failed_requests := current_state.requests_failed - - (self._previous_crawler_state or current_state).requests_failed + - (self._run.previous_state or current_state).requests_failed ) > 0: message = f'Experiencing problems, {failed_requests} failed requests since last status update.' else: @@ -1937,7 +1958,7 @@ async def _crawler_state_task(self) -> None: ) if self._status_message_callback: - new_message = await self._status_message_callback(current_state, self._previous_crawler_state, message) + new_message = await self._status_message_callback(current_state, self._run.previous_state, message) if new_message: message = new_message self._log_status_message(message, level='INFO') @@ -1948,7 +1969,7 @@ async def _crawler_state_task(self) -> None: event=Event.CRAWLER_STATUS, event_data=EventCrawlerStatusData(message=message, crawler_id=self._id) ) - self._previous_crawler_state = current_state + self._run.previous_state = current_state async def _mark_request_as_handled(self, request: Request) -> None: request_manager = await self.get_request_manager() @@ -1960,4 +1981,4 @@ async def _mark_request_as_handled(self, request: Request) -> None: logger=self._logger, max_retries=3, ) - self._handled_request_this_run = True + self._run.handled_request = True diff --git a/src/crawlee/crawlers/_basic/_crawler_run.py b/src/crawlee/crawlers/_basic/_crawler_run.py new file mode 100644 index 0000000000..e3c5dae5ed --- /dev/null +++ b/src/crawlee/crawlers/_basic/_crawler_run.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Generic + +from typing_extensions import TypeVar + +from crawlee._utils.log import LoggerOnce +from crawlee.autoscaling import AutoscaledPool +from crawlee.statistics import StatisticsState + +if TYPE_CHECKING: + import logging + from collections.abc import Awaitable, Callable + + from crawlee.autoscaling import ConcurrencyConsumer + from crawlee.autoscaling._types import ConcurrencySystemInterface + +TStatisticsState = TypeVar('TStatisticsState', bound=StatisticsState, default=StatisticsState) + + +class CrawlerRun(Generic[TStatisticsState]): + """Everything that belongs to a single `BasicCrawler.run` call. + + The crawler builds a fresh instance for every run, so nothing has to be reset between runs. A finished run stays + readable as the history of the last crawl, so liveness is asked about explicitly with `is_live`. + """ + + def __init__( + self, + *, + logger: logging.Logger, + concurrency_system: ConcurrencySystemInterface, + consumer: ConcurrencyConsumer, + run_task_function: Callable[[str], Awaitable], + is_task_ready_function: Callable[[], Awaitable[bool]], + is_finished_function: Callable[[], Awaitable[bool]], + ) -> None: + self._autoscaled_pool = AutoscaledPool( + concurrency_system=concurrency_system, + consumer=consumer, + run_task_function=run_task_function, + is_task_ready_function=is_task_ready_function, + is_finished_function=is_finished_function, + ) + self._logger_once = LoggerOnce(logger) + self._stop_requested = False + self._finished = False + + self.failed = False + """Whether a request has failed while the crawler aborts on an error.""" + + self.handled_request = False + """Whether this run has marked any request as handled.""" + + self.previous_state: TStatisticsState | None = None + """The statistics state at the last status update of this run.""" + + @property + def is_live(self) -> bool: + """Whether the `BasicCrawler.run` call this belongs to is still executing.""" + return not self._finished + + @property + def stop_requested(self) -> bool: + """Whether a graceful shutdown of this run has been requested.""" + return self._stop_requested + + async def dispatch_requests(self) -> None: + """Run the task loop until the crawl is over. The concurrency system has to be active by now.""" + await self._autoscaled_pool.run() + + def stop(self) -> None: + """Ask the run to stop taking new requests.""" + self._stop_requested = True + + def log_once(self, message: str, key: str) -> None: + """Log `message` the first time this run asks for it under `key`.""" + self._logger_once.log(message, key=key) + + def finish(self) -> None: + """Mark the run as history, so that the crawler can start another one.""" + self._finished = True diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index f4b614b557..c5cfbbca36 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -1925,6 +1925,36 @@ async def test_stop_before_run_starts() -> None: await asyncio.wait_for(run_task, timeout=10) +async def test_stop_between_runs() -> None: + """A stop called between runs ends the next run and doesn't affect the one after it.""" + handler = AsyncMock() + crawler = BasicCrawler(request_handler=handler) + await crawler.run() + + crawler.stop() + await crawler.run(['https://a.placeholder.com']) + handler.assert_not_called() + + await crawler.run() + handler.assert_called_once() + + +async def test_run_while_running() -> None: + """A second `run` call raises and leaves the run in progress going.""" + crawler = BasicCrawler(keep_alive=True, request_handler=AsyncMock()) + + run_task = asyncio.create_task(crawler.run()) + await asyncio.sleep(0) + + with pytest.raises(RuntimeError, match='already running'): + await crawler.run() + + assert not run_task.done() + + crawler.stop() + await asyncio.wait_for(run_task, timeout=10) + + async def test_services_no_side_effect_on_crawler_init() -> None: custom_configuration = Configuration() custom_event_manager = LocalEventManager.from_config(custom_configuration) @@ -2497,7 +2527,7 @@ async def status_callback( return None crawler = BasicCrawler(status_message_callback=status_callback) - crawler._previous_crawler_state = StatisticsState(requests_failed=2) + crawler._last_run = Mock(previous_state=StatisticsState(requests_failed=2)) crawler._statistics = Mock(state=StatisticsState(requests_failed=5)) async with service_locator.get_event_manager(): @@ -2507,6 +2537,28 @@ async def status_callback( assert problem_messages == ['Experiencing problems, 3 failed requests since last status update.'] +async def test_status_callback_previous_state_per_run() -> None: + """The first status update of a repeated run has no previous state.""" + status_updated = asyncio.Event() + status_callback = AsyncMock(side_effect=lambda *_: status_updated.set()) + + async def handler(context: BasicCrawlingContext) -> None: + await status_updated.wait() + status_updated.clear() + + crawler = BasicCrawler( + request_handler=handler, + status_message_callback=status_callback, + status_message_logging_interval=timedelta(hours=1), + ) + + await crawler.run(['https://a.placeholder.com']) + await crawler.run(['https://b.placeholder.com']) + + previous_states = [call.args[1] for call in status_callback.call_args_list] + assert previous_states == [None, None] + + @pytest.mark.parametrize( ('queue_name', 'queue_alias', 'by_id'), [ From fce44d595f3773a2b40c8b9dcc4e11941131bcb6 Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Mon, 2 Nov 2026 14:06:56 +0200 Subject: [PATCH 2/3] Apply batched suggestions from code review Co-authored-by: Vlada Dusek --- src/crawlee/crawlers/_basic/_basic_crawler.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index fec42e5c92..cb3056db62 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -571,7 +571,7 @@ def statistics(self) -> Statistics[TStatisticsState]: @property def _run(self) -> CrawlerRun[TStatisticsState]: - """The last run started on this instance, for the code that only a run drives.""" + """The last run started on this instance. Only code that a run drives may use it.""" if self._last_run is None: raise RuntimeError('No crawl has been started yet, only `run()` starts one.') @@ -595,7 +595,7 @@ def stop(self, reason: str = 'Stop was called externally.') -> None: if self._live_run is not None: self._live_run.stop() else: - # `create_task(crawler.run())` doesn't start the coroutine yet, so the stop waits for the next run. + # A caller can stop a run it scheduled with `create_task(crawler.run())` before the task starts. self._stop_before_run = True def _wrap_handler_with_error_context( From f2355d37329c9ef4fd41bb4b035a49b3fa30105b Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Mon, 2 Nov 2026 12:11:06 +0000 Subject: [PATCH 3/3] update docsting --- src/crawlee/crawlers/_basic/_basic_crawler.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index cb3056db62..b7018144ad 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -585,7 +585,8 @@ def _live_run(self) -> CrawlerRun[TStatisticsState] | None: def stop(self, reason: str = 'Stop was called externally.') -> None: """Set flag to stop crawler. - This stops current crawler run regardless of whether all requests were finished. + This stops current crawler run regardless of whether all requests were finished. Called with no run in progress, + it stops the next run. Args: reason: Reason for stopping that will be used in logs.