Repository: apify/crawlee-python
File: src/crawlee/_autoscaling/snapshotter.py (Python)

Code before:
        self._active = False

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

        Args:
            config: The configuration object. Uses the global (default) configuration if not provided.
        """
        config = service_locator.get_configuration()
        max_memory_size = cls._determine_max_memory_size(config.memory_mbytes, config.available_memory_ratio)

        return cls(
            max_used_cpu_ratio=config.max_used_cpu_ratio,
            max_used_memory_ratio=config.max_used_memory_ratio,
            max_event_loop_delay=config.max_event_loop_delay,
            max_client_errors=config.max_client_errors,
            max_memory_size=max_memory_size,
        )

    @staticmethod
    def _get_sorted_list_by_created_at(input_list: list[T]) -> SortedList[T]:
        return SortedList(input_list, key=attrgetter('created_at'))

    @staticmethod
    def _determine_max_memory_size(memory_mbytes: int | None, available_memory_ratio: float) -> ByteSize:
        """Determine the maximum memory size for the current run.

        If `memory_mbytes` is provided, it uses that value. Otherwise, it calculates the default `max_memory_size`
        as a proportion of the system's total available memory based on `available_memory_ratio`.
        """
        if memory_mbytes:
            return ByteSize.from_mb(memory_mbytes)

        max_memory_size_in_bytes = int(psutil.virtual_memory().total * available_memory_ratio)
        max_memory_size = ByteSize(max_memory_size_in_bytes)
        logger.info(f'Setting max_memory_size of this run to {max_memory_size}.')
        return max_memory_size

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

    async def __aenter__(self) -> Snapshotter:
        """Starts capturing snapshots at configured intervals.

        Raises:
            RuntimeError: If the context manager is already active.
        """
        if self._active:

Review comment (suggestion):
I wouldn't mind inlining this into `from_config` :slightly_smiling_face: But I'll leave that up to you

Code after:
        self._active = False

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

        Args:
            config: The configuration object. Uses the global (default) configuration if not provided.
        """
        config = service_locator.get_configuration()

        # Compute the maximum memory size based on the provided configuration. If `memory_mbytes` is provided,
        # it uses that value. Otherwise, it calculates the `max_memory_size` as a proportion of the system's
        # total available memory based on `available_memory_ratio`.
        max_memory_size = (
            ByteSize.from_mb(config.memory_mbytes)
            if config.memory_mbytes
            else ByteSize(int(get_memory_info().total_size.bytes * config.available_memory_ratio))
        )

        return cls(
            max_used_cpu_ratio=config.max_used_cpu_ratio,
            max_used_memory_ratio=config.max_used_memory_ratio,
            max_event_loop_delay=config.max_event_loop_delay,
            max_client_errors=config.max_client_errors,
            max_memory_size=max_memory_size,
        )

    @staticmethod
    def _get_sorted_list_by_created_at(input_list: list[T]) -> SortedList[T]:
        return SortedList(input_list, key=attrgetter('created_at'))

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

    async def __aenter__(self) -> Snapshotter:
        """Starts capturing snapshots at configured intervals.

        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
        event_manager = service_locator.get_event_manager()
        event_manager.on(event=Event.SYSTEM_INFO, listener=self._snapshot_cpu)
        event_manager.on(event=Event.SYSTEM_INFO, listener=self._snapshot_memory)
        self._snapshot_event_loop_task.start()