Repository: apify/crawlee-python
File: src/crawlee/events/_event_manager.py (Python)

Code before:
            func=self._emit_persist_state_event,
            delay=self._persist_state_interval,
        )

        # Flag to indicate the context state.
        self._initialized = False

    async def __aenter__(self) -> EventManager:
        """Initializes the event manager upon entering the async context."""
        if not self._initialized:
            self._emit_persist_state_event_rec_task.start()
            self._initialized = True

        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        exc_traceback: TracebackType | None,
    ) -> None:
        """Closes the local event manager upon exiting the async context.

        This will stop listening for the events, and it will wait for all the event listeners to finish.
        """
        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

    def on(self, *, event: Event, listener: Listener) -> None:
        """Add an event listener to the event manager.

        Args:
            event: The Actor event for which to listen to.
            listener: The function (sync or async) which is to be called when the event is emitted.
        """

        @wraps(listener)
        async def listener_wrapper(event_data: EventData) -> None:
            # If the listener is a coroutine function, just call it, otherwise, run it in a separate thread
            # to avoid blocking the event loop
            coro = (
                listener(event_data)
                if asyncio.iscoroutinefunction(listener)
                else asyncio.to_thread(listener, event_data)
            )
            # Note: use `asyncio.iscoroutinefunction` rather then `inspect.iscoroutinefunction` since it works with

Review comment (suggestion):
This is kind of tricky when thinking about usecase (maybe irrelevant)

```
async with event_manager:
  # some code running in context. As expected.
  async with event_manager:
    # some code running in context. As expected.
  # some code no longer running in actual context, even though it is still in the outer context. Might be surprising.
```

Code after:
        self._emit_persist_state_event_rec_task = RecurringTask(
            func=self._emit_persist_state_event,
            delay=self._persist_state_interval,
        )

        # Flag to indicate the context state.
        self._active = False

    @property
    def active(self) -> bool:
        """Indicates whether the context is active."""
        return self._active

    async def __aenter__(self) -> EventManager:
        """Initializes the event manager upon entering the async context.

        Raises:
            RuntimeError: If the context manager is already active.
        """
        if self._active:
            raise RuntimeError(f'The {self.__class__.__name__} is already active.')

        self._active = True
        self._emit_persist_state_event_rec_task.start()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        exc_traceback: TracebackType | None,
    ) -> None:
        """Closes the local event manager upon exiting the async context.

        This will stop listening for the events, and it will wait for all the event listeners to finish.

        Raises:
            RuntimeError: If the context manager is not active.
        """
        if not self._active:
            raise RuntimeError(f'The {self.__class__.__name__} is not active.')

        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._active = False

    def on(self, *, event: Event, listener: Listener) -> None:
        """Add an event listener to the event manager.