Repository: apify/crawlee-python
File: src/crawlee/crawlers/_basic/_basic_crawler.py (Python)

Code before:
        event_manager = self._service_locator.get_event_manager()

        # 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 = [
            cm
            for cm in (
                event_manager,
                self._snapshotter,
                self._statistics,
                self._session_pool if self._use_session_pool else None,
                self._http_client,
                *self._additional_context_managers,
            )
            if cm and getattr(cm, 'active', False) is False
        ]

        async with AsyncExitStack() as exit_stack:
            for context in contexts_to_enter:
                await exit_stack.enter_async_context(context)  # type: ignore[arg-type]

            self._crawler_state_rec_task.start()
            try:
                await self._autoscaled_pool.run()
            finally:
                await self._crawler_state_rec_task.stop()

            # Emit PERSIST_STATE event when crawler is finishing to allow listeners to persist their state if needed
            if not self.statistics.state.crawler_last_started_at:
                raise RuntimeError('Statistics.state.crawler_last_started_at not set.')
            run_duration = datetime.now(timezone.utc) - self.statistics.state.crawler_last_started_at
            self._statistics.state.crawler_runtime = self.statistics.state.crawler_runtime + run_duration
            self._service_locator.get_event_manager().emit(
                event=Event.PERSIST_STATE, event_data=EventPersistStateData(is_migrating=False)
            )

    async def add_requests(
        self,
        requests: Sequence[str | Request],
        *,
        forefront: bool = False,
        batch_size: int = 1000,
        wait_time_between_batches: timedelta = timedelta(0),
        wait_for_all_requests_to_be_added: bool = False,
        wait_for_all_requests_to_be_added_timeout: timedelta | None = None,
    ) -> None:
        """Add requests to the underlying request manager in batches.

        Args:
            requests: A list of requests to add to the queue.
            forefront: If True, add requests to the forefront of the queue.

Review comment (suggestion):
Maybe `RecurringTask` could also be an async context manager (two options of usage - `start`/`stop` and `aenter`/`aexit`) and then it can be just another `contexts_to_enter`.

Code after:
        final_statistics = self._statistics.calculate()
        if self._statistics_log_format == 'table':
            self._logger.info(f'Final request statistics:\n{final_statistics.to_table()}')
        else:
            self._logger.info('Final request statistics:', extra=final_statistics.to_dict())
        return final_statistics

    async def _run_crawler(self) -> None:
        event_manager = self._service_locator.get_event_manager()

        # 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 = [
            cm
            for cm in (
                event_manager,
                self._snapshotter,
                self._statistics,
                self._session_pool if self._use_session_pool else None,
                self._http_client,
                self._crawler_state_rec_task,
                *self._additional_context_managers,
            )
            if cm and getattr(cm, 'active', False) is False
        ]

        async with AsyncExitStack() as exit_stack:
            for context in contexts_to_enter:
                await exit_stack.enter_async_context(context)  # type: ignore[arg-type]

            await self._autoscaled_pool.run()

            # Emit PERSIST_STATE event when crawler is finishing to allow listeners to persist their state if needed
            event_manager.emit(event=Event.PERSIST_STATE, event_data=EventPersistStateData(is_migrating=False))

    async def add_requests(
        self,
        requests: Sequence[str | Request],
        *,
        forefront: bool = False,
        batch_size: int = 1000,
        wait_time_between_batches: timedelta = timedelta(0),
        wait_for_all_requests_to_be_added: bool = False,
        wait_for_all_requests_to_be_added_timeout: timedelta | None = None,
    ) -> None:
        """Add requests to the underlying request manager in batches.

        Args:
            requests: A list of requests to add to the queue.
            forefront: If True, add requests to the forefront of the queue.
            batch_size: The number of requests to add in one batch.