code quality up

Author: Mantisus

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -61,7 +61,6 @@
     UserDefinedErrorHandlerError,
     UserHandlerTimeoutError,
 )
-from crawlee.events import EventManager
 from crawlee.events._types import Event, EventCrawlerStatusData
 from crawlee.http_clients import ImpitHttpClient
 from crawlee.router import Router
@@ -91,6 +90,7 @@
         PushDataKwargs,
     )
     from crawlee.configuration import Configuration
+    from crawlee.events import EventManager
     from crawlee.http_clients import HttpClient, HttpResponse
     from crawlee.proxy_configuration import ProxyConfiguration, ProxyInfo
     from crawlee.request_loaders import RequestManager
@@ -773,26 +773,31 @@ async def _run_crawler(self) -> None:
         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]
+        )
+
         # 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 (
-                global_event_manager,
-                local_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 (isinstance(cm, EventManager) or not getattr(cm, 'active', False))
-        ]
+        contexts_to_enter.extend(
+            [
+                cm
+                for cm in (
+                    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)  # ty: ignore[invalid-argument-type]
+                await exit_stack.enter_async_context(context)
 
             await self._autoscaled_pool.run()
 

src/crawlee/events/_event_manager.py

@@ -94,24 +94,19 @@ def __init__(
         )
 
         # Reference count for active contexts.
-        self._ref_count = 0
-
-        # Flag to indicate the context state.
-        self._active = False
+        self._active_ref_count = 0
 
     @property
     def active(self) -> bool:
         """Indicate whether the context is active."""
-        return self._active
+        return self._active_ref_count > 0
 
     async def __aenter__(self) -> EventManager:
         """Initialize the event manager upon entering the async context."""
-        self._ref_count += 1
-
-        if self._ref_count > 1:
+        self._active_ref_count += 1
+        if self._active_ref_count > 1:
             return self
 
-        self._active = True
         self._emit_persist_state_event_rec_task.start()
         return self
 
@@ -128,24 +123,23 @@ async def __aexit__(
         Raises:
             RuntimeError: If the context manager is not active.
         """
-        if not self._active:
+        if not self.active:
             raise RuntimeError(f'The {self.__class__.__name__} is not active.')
 
-        self._ref_count -= 1
-
-        # Emit persist state event to ensure the latest state is saved before closing the context.
-        await self._emit_persist_state_event()
-
-        if self._ref_count > 0:
+        if self._active_ref_count > 1:
+            # Emit persist state event to ensure the latest state is saved before closing the context.
+            await self._emit_persist_state_event()
+            self._active_ref_count -= 1
             return
 
         # Stop persist state event periodic emission and manually emit last one to ensure latest state is saved.
         await self._emit_persist_state_event_rec_task.stop()
+        await self._emit_persist_state_event()
         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()
-        self._active = False
+        self._active_ref_count -= 1
 
     @overload
     def on(self, *, event: Literal[Event.PERSIST_STATE], listener: EventListener[EventPersistStateData]) -> None: ...