Class LocalEventManager

src/crawlee/events/_local_event_manager.py:26–105

Event manager for local environments. It extends the `EventManager` to emit `SystemInfo` events at regular intervals. The `LocalEventManager` is intended to be used in local environments, where the system metrics are required managing the `Snapshotter` and `AutoscaledPool`.

@docs_group('Event managers')
class LocalEventManager(EventManager):
    """Event manager for local environments.

    It extends the `EventManager` to emit `SystemInfo` events at regular intervals. The `LocalEventManager`
    is intended to be used in local environments, where the system metrics are required managing the `Snapshotter`
    and `AutoscaledPool`.
    """

    def __init__(
        self,
        system_info_interval: timedelta = timedelta(seconds=1),
        **event_manager_options: Unpack[EventManagerOptions],
    ) -> None:
        """Initialize a new instance.

        In most cases, you should use the `from_config` constructor to create a new instance based on
        the provided configuration.

        Args:
            system_info_interval: Interval at which `SystemInfo` events are emitted.
            event_manager_options: Additional options for the parent class.
        """
        self._system_info_interval = system_info_interval

        # Recurring task for emitting system info events.
        self._emit_system_info_event_rec_task = RecurringTask(
            func=self._emit_system_info_event,
            delay=self._system_info_interval,
        )

        super().__init__(**event_manager_options)

    @classmethod
    def from_config(cls, config: Configuration | None = None) -> LocalEventManager:
        """Initialize a new instance based on the provided `Configuration`.

        Args:
            config: The `Configuration` instance. Uses the global (default) one if not provided.
        """
        config = config or Configuration.get_global_configuration()

        return cls(
            system_info_interval=config.system_info_interval,
            persist_state_interval=config.persist_state_interval,
        )

    async def __aenter__(self) -> LocalEventManager:
        """Initialize the local event manager upon entering the async context.

        It starts emitting system info events at regular intervals.
        """
        await super().__aenter__()

        if self._active_ref_count == 1:
            self._emit_system_info_event_rec_task.start()

        return self