Fix cycle imports, tests and everything

Author: vdusek

src/crawlee/_autoscaling/snapshotter.py

@@ -20,6 +20,8 @@
 if TYPE_CHECKING:
     from types import TracebackType
 
+    from crawlee.events import EventManager
+
 logger = getLogger(__name__)
 
 T = TypeVar('T')
@@ -38,6 +40,7 @@ class Snapshotter:
     def __init__(
         self,
         *,
+        event_manager: EventManager | None = None,
         event_loop_snapshot_interval: timedelta = timedelta(milliseconds=500),
         client_snapshot_interval: timedelta = timedelta(milliseconds=1000),
         max_used_cpu_ratio: float = 0.95,
@@ -54,6 +57,8 @@ def __init__(
         """A default constructor.
 
         Args:
+            event_manager: The event manager used to emit system info events. From data provided by this event
+              the CPU and memory usage are read.
             event_loop_snapshot_interval: The interval at which the event loop is sampled.
             client_snapshot_interval: The interval at which the client is sampled.
             max_used_cpu_ratio: Sets the ratio, defining the maximum CPU usage. When the CPU usage is higher than
@@ -79,8 +84,7 @@ def __init__(
         if available_memory_ratio is None and max_memory_size is None:
             raise ValueError('At least one of `available_memory_ratio` or `max_memory_size` must be specified')
 
-        self._event_manager = service_container.get_event_manager()
-
+        self._event_manager = event_manager or service_container.get_event_manager()
         self._event_loop_snapshot_interval = event_loop_snapshot_interval
         self._client_snapshot_interval = client_snapshot_interval
         self._max_event_loop_delay = max_event_loop_delay

src/crawlee/basic_crawler/_basic_crawler.py

@@ -50,6 +50,8 @@
 
     from crawlee._types import ConcurrencySettings, HttpMethod, JsonSerializable
     from crawlee.base_storage_client._models import DatasetItemsListPage
+    from crawlee.configuration import Configuration
+    from crawlee.events._event_manager import EventManager
     from crawlee.http_clients import BaseHttpClient, HttpResponse
     from crawlee.proxy_configuration import ProxyConfiguration, ProxyInfo
     from crawlee.sessions import Session
@@ -93,6 +95,9 @@ class BasicCrawlerOptions(TypedDict, Generic[TCrawlingContext]):
     """Maximum number of session rotations per request. The crawler rotates the session if a proxy error occurs
     or if the website blocks the request."""
 
+    configuration: NotRequired[Configuration]
+    """Crawler configuration."""
+
     request_handler_timeout: NotRequired[timedelta]
     """Maximum duration allowed for a single request handler to run."""
 
@@ -111,6 +116,9 @@ class BasicCrawlerOptions(TypedDict, Generic[TCrawlingContext]):
     statistics: NotRequired[Statistics[StatisticsState]]
     """A custom `Statistics` instance, allowing the use of non-default configuration."""
 
+    event_manager: NotRequired[EventManager]
+    """A custom `EventManager` instance, allowing the use of non-default configuration."""
+
     configure_logging: NotRequired[bool]
     """If True, the crawler will set up logging infrastructure automatically."""
 
@@ -157,126 +165,136 @@ class BasicCrawler(Generic[TCrawlingContext]):
     def __init__(
         self,
         *,
+        configuration: Configuration | None = None,
+        event_manager: EventManager | None = None,
         request_provider: RequestProvider | None = None,
-        request_handler: Callable[[TCrawlingContext], Awaitable[None]] | None = None,
+        session_pool: SessionPool | None = None,
+        proxy_configuration: ProxyConfiguration | None = None,
         http_client: BaseHttpClient | None = None,
-        concurrency_settings: ConcurrencySettings | None = None,
+        request_handler: Callable[[TCrawlingContext], Awaitable[None]] | None = None,
         max_request_retries: int = 3,
         max_requests_per_crawl: int | None = None,
         max_session_rotations: int = 10,
-        request_handler_timeout: timedelta = timedelta(minutes=1),
-        session_pool: SessionPool | None = None,
+        max_crawl_depth: int | None = None,
         use_session_pool: bool = True,
         retry_on_blocked: bool = True,
-        proxy_configuration: ProxyConfiguration | None = None,
+        concurrency_settings: ConcurrencySettings | None = None,
+        request_handler_timeout: timedelta = timedelta(minutes=1),
         statistics: Statistics | None = None,
         configure_logging: bool = True,
-        max_crawl_depth: int | None = None,
         _context_pipeline: ContextPipeline[TCrawlingContext] | None = None,
         _additional_context_managers: Sequence[AsyncContextManager] | None = None,
         _logger: logging.Logger | None = None,
     ) -> None:
         """A default constructor.
 
         Args:
+            configuration: The configuration object. Some of its properties are used as defaults for the crawler.
+            event_manager: The event manager for managing events for the crawler and all its components.
             request_provider: Provider for requests to be processed by the crawler.
-            request_handler: A callable responsible for handling requests.
+            session_pool: A custom `SessionPool` instance, allowing the use of non-default configuration.
+            proxy_configuration: HTTP proxy configuration used when making requests.
             http_client: HTTP client used by `BasicCrawlingContext.send_request` and the HTTP-based crawling.
-            concurrency_settings: Settings to fine-tune concurrency levels.
+            request_handler: A callable responsible for handling requests.
             max_request_retries: Maximum number of attempts to process a single request.
             max_requests_per_crawl: Maximum number of pages to open during a crawl. The crawl stops upon reaching
                 this limit. Setting this value can help avoid infinite loops in misconfigured crawlers. `None` means
                 no limit. Due to concurrency settings, the actual number of pages visited may slightly exceed
                 this value.
             max_session_rotations: Maximum number of session rotations per request. The crawler rotates the session
                 if a proxy error occurs or if the website blocks the request.
-            request_handler_timeout: Maximum duration allowed for a single request handler to run.
+            max_crawl_depth: Maximum crawl depth. If set, the crawler will stop crawling after reaching this depth.
             use_session_pool: Enable the use of a session pool for managing sessions during crawling.
-            session_pool: A custom `SessionPool` instance, allowing the use of non-default configuration.
             retry_on_blocked: If True, the crawler attempts to bypass bot protections automatically.
-            proxy_configuration: HTTP proxy configuration used when making requests.
+            concurrency_settings: Settings to fine-tune concurrency levels.
+            request_handler_timeout: Maximum duration allowed for a single request handler to run.
             statistics: A custom `Statistics` instance, allowing the use of non-default configuration.
             configure_logging: If True, the crawler will set up logging infrastructure automatically.
-            max_crawl_depth: Maximum crawl depth. If set, the crawler will stop crawling after reaching this depth.
             _context_pipeline: Enables extending the request lifecycle and modifying the crawling context.
                 Intended for use by subclasses rather than direct instantiation of `BasicCrawler`.
             _additional_context_managers: Additional context managers used throughout the crawler lifecycle.
             _logger: A logger instance, typically provided by a subclass, for consistent logging labels.
         """
-        self._router: Router[TCrawlingContext] | None = None
+        if configuration:
+            service_container.set_configuration(configuration)
+        if event_manager:
+            service_container.set_event_manager(event_manager)
+
+        config = service_container.get_configuration()
+
+        # Core components
+        self._request_provider = request_provider
+        self._session_pool = session_pool or SessionPool()
+        self._proxy_configuration = proxy_configuration
+        self._http_client = http_client or HttpxHttpClient()
 
+        # Request router setup
+        self._router: Router[TCrawlingContext] | None = None
         if isinstance(cast(Router, request_handler), Router):
             self._router = cast(Router[TCrawlingContext], request_handler)
         elif request_handler is not None:
             self._router = None
             self.router.default_handler(request_handler)
 
-        self._http_client = http_client or HttpxHttpClient()
-
-        self._context_pipeline = (_context_pipeline or ContextPipeline()).compose(self._check_url_after_redirects)
-
+        # Error & failed request handlers
         self._error_handler: ErrorHandler[TCrawlingContext | BasicCrawlingContext] | None = None
         self._failed_request_handler: FailedRequestHandler[TCrawlingContext | BasicCrawlingContext] | None = None
 
+        # Context pipeline
+        self._context_pipeline = (_context_pipeline or ContextPipeline()).compose(self._check_url_after_redirects)
+
+        # Crawl settings
         self._max_request_retries = max_request_retries
         self._max_requests_per_crawl = max_requests_per_crawl
         self._max_session_rotations = max_session_rotations
+        self._max_crawl_depth = max_crawl_depth
 
-        self._request_provider = request_provider
-
-        config = service_container.get_configuration()
-
+        # Timeouts
         self._request_handler_timeout = request_handler_timeout
         self._internal_timeout = (
             config.internal_timeout
             if config.internal_timeout is not None
             else max(2 * request_handler_timeout, timedelta(minutes=5))
         )
 
-        self._tld_extractor = TLDExtract(cache_dir=tempfile.TemporaryDirectory().name)
-
-        self._event_manager = service_container.get_event_manager()
-        self._snapshotter = Snapshotter(
-            max_memory_size=ByteSize.from_mb(config.memory_mbytes) if config.memory_mbytes else None,
-            available_memory_ratio=config.available_memory_ratio,
-        )
-        self._pool = AutoscaledPool(
-            system_status=SystemStatus(self._snapshotter),
-            is_finished_function=self.__is_finished_function,
-            is_task_ready_function=self.__is_task_ready_function,
-            run_task_function=self.__run_task_function,
-            concurrency_settings=concurrency_settings,
-        )
-
+        # Retry and session settings
         self._use_session_pool = use_session_pool
-        self._session_pool = session_pool or SessionPool()
-
         self._retry_on_blocked = retry_on_blocked
 
+        # Logging setup
         if configure_logging:
             root_logger = logging.getLogger()
             configure_logger(root_logger, remove_old_handlers=True)
-
-            # Silence HTTPX logger
-            httpx_logger = logging.getLogger('httpx')
+            httpx_logger = logging.getLogger('httpx')  # Silence HTTPX logger
             httpx_logger.setLevel(logging.DEBUG if get_configured_log_level() <= logging.DEBUG else logging.WARNING)
+        self._logger = _logger or logging.getLogger(__name__)
 
-        if not _logger:
-            _logger = logging.getLogger(__name__)
-
-        self._logger = _logger
-
-        self._proxy_configuration = proxy_configuration
+        # Statistics
         self._statistics = statistics or Statistics(
-            event_manager=self._event_manager,
             periodic_message_logger=self._logger,
             log_message='Current request statistics:',
         )
+
+        # Additional context managers to enter and exit
         self._additional_context_managers = _additional_context_managers or []
 
+        # Internal, not explicitly configurable components
+        self._tld_extractor = TLDExtract(cache_dir=tempfile.TemporaryDirectory().name)
+        self._snapshotter = Snapshotter(
+            max_memory_size=ByteSize.from_mb(config.memory_mbytes) if config.memory_mbytes else None,
+            available_memory_ratio=config.available_memory_ratio,
+        )
+        self._autoscaled_pool = AutoscaledPool(
+            system_status=SystemStatus(self._snapshotter),
+            is_finished_function=self.__is_finished_function,
+            is_task_ready_function=self.__is_task_ready_function,
+            run_task_function=self.__run_task_function,
+            concurrency_settings=concurrency_settings,
+        )
+
+        # State flags
         self._running = False
         self._has_finished_before = False
-        self._max_crawl_depth = max_crawl_depth
 
     @property
     def log(self) -> logging.Logger:
@@ -459,8 +477,10 @@ def sigint_handler() -> None:
         return final_statistics
 
     async def _run_crawler(self) -> None:
+        event_manager = service_container.get_event_manager()
+
         async with AsyncExitStack() as exit_stack:
-            await exit_stack.enter_async_context(self._event_manager)
+            await exit_stack.enter_async_context(event_manager)
             await exit_stack.enter_async_context(self._snapshotter)
             await exit_stack.enter_async_context(self._statistics)
 
@@ -470,7 +490,7 @@ async def _run_crawler(self) -> None:
             for context_manager in self._additional_context_managers:
                 await exit_stack.enter_async_context(context_manager)
 
-            await self._pool.run()
+            await self._autoscaled_pool.run()
 
     async def add_requests(
         self,

src/crawlee/configuration.py

@@ -5,9 +5,7 @@
 
 from pydantic import AliasChoices, BeforeValidator, Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
-from typing_extensions import Self
 
-from crawlee import service_container
 from crawlee._utils.docs import docs_group
 from crawlee._utils.models import timedelta_ms
 
@@ -236,11 +234,13 @@ class Configuration(BaseSettings):
     def get_global_configuration(cls) -> Self:
         """Retrieve the global instance of the configuration.
 
-        Mostly for the backward compatibility.
+        TODO: Can we remove this?
         """
-        config = service_container.get_configuration()
+        from crawlee.service_container import get_configuration
 
-        if not isinstance(config, cls):
-            raise TypeError(f'Requested configuration of type {cls}, but got {config.__class__} instead.')
+        cfg = get_configuration()
 
-        return config
+        if not isinstance(cfg, cls):
+            raise TypeError(f'Requested global configuration object of type {cls}, but {cfg.__class__} was found')
+
+        return cfg

src/crawlee/events/_event_manager.py

@@ -80,9 +80,21 @@ def __init__(
             delay=self._persist_state_interval,
         )
 
+        self._initialized = False
+
+    @property
+    def initialized(self) -> bool:
+        """Indicates whether the event manager has been initialized."""
+        return self._initialized
+
     async def __aenter__(self) -> EventManager:
         """Initializes the event manager upon entering the async context."""
-        self._emit_persist_state_event_rec_task.start()
+        if not self._initialized:
+            self._emit_persist_state_event_rec_task.start()
+            self._initialized = True
+        else:
+            logger.warning('Calling EventManager.__aenter__(), but event manager was already initialized. Skipping...')
+
         return self
 
     async def __aexit__(
@@ -95,11 +107,15 @@ async def __aexit__(
 
         This will stop listening for the events, and it will wait for all the event listeners to finish.
         """
-        await self.wait_for_all_listeners_to_complete(timeout=self._close_timeout)
-        self._event_emitter.remove_all_listeners()
-        self._listener_tasks.clear()
-        self._listeners_to_wrappers.clear()
-        await self._emit_persist_state_event_rec_task.stop()
+        if self._initialized:
+            await self.wait_for_all_listeners_to_complete(timeout=self._close_timeout)
+            self._event_emitter.remove_all_listeners()
+            self._listener_tasks.clear()
+            self._listeners_to_wrappers.clear()
+            await self._emit_persist_state_event_rec_task.stop()
+            self._initialized = False
+        else:
+            logger.warning('Calling EventManager.__aexit__(), but event manager was not yet initialized. Skipping...')
 
     def on(self, *, event: Event, listener: Listener) -> None:
         """Add an event listener to the event manager.
@@ -125,17 +141,17 @@ async def listener_wrapper(event_data: EventData) -> None:
             self._listener_tasks.add(listener_task)
 
             try:
-                logger.debug('LocalEventManager.on.listener_wrapper(): Awaiting listener task...')
+                logger.debug('EventManager.on.listener_wrapper(): Awaiting listener task...')
                 await listener_task
-                logger.debug('LocalEventManager.on.listener_wrapper(): Listener task completed.')
+                logger.debug('EventManager.on.listener_wrapper(): Listener task completed.')
             except Exception:
                 # We need to swallow the exception and just log it here, otherwise it could break the event emitter
                 logger.exception(
                     'Exception in the event listener',
                     extra={'event_name': event.value, 'listener_name': listener.__name__},
                 )
             finally:
-                logger.debug('LocalEventManager.on.listener_wrapper(): Removing listener task from the set...')
+                logger.debug('EventManager.on.listener_wrapper(): Removing listener task from the set...')
                 self._listener_tasks.remove(listener_task)
 
         self._listeners_to_wrappers[event][listener].append(listener_wrapper)
@@ -164,6 +180,9 @@ def emit(self, *, event: Event, event_data: EventData) -> None:
             event: The event which will be emitted.
             event_data: The data which will be passed to the event listeners.
         """
+        if not self._initialized:
+            raise RuntimeError('EventManager is not initialized. Please use it within async context manager.')
+
         self._event_emitter.emit(event.value, event_data)
 
     async def wait_for_all_listeners_to_complete(self, *, timeout: timedelta | None = None) -> None:
@@ -173,6 +192,8 @@ async def wait_for_all_listeners_to_complete(self, *, timeout: timedelta | None
             timeout: The maximum time to wait for the event listeners to finish. If they do not complete within
                 the specified timeout, they will be canceled.
         """
+        if not self._initialized:
+            raise RuntimeError('EventManager is not initialized. Please use it within async context manager.')
 
         async def wait_for_listeners() -> None:
             """Gathers all listener tasks and awaits their completion, logging any exceptions encountered."""