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

Code before:
    """The maximum ratio of CPU that is considered acceptable."""

    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    """The time at which the system load information was measured."""

    @property
    def is_overloaded(self) -> bool:
        """Indicate whether the CPU is considered as overloaded."""
        return self.used_ratio > self.max_used_ratio


@dataclass
class MemorySnapshot:
    """A snapshot of memory usage."""

    current_size: ByteSize
    """Memory usage of the current Python process and its children."""

    system_wide_used_size: ByteSize | None
    """Memory usage of all processes, system-wide."""

    max_memory_size: ByteSize | Ratio
    """The maximum memory that can be used by `AutoscaledPool`.

    When of type `ByteSize` then it is used as fixed memory size. When of type `Ratio` then it allows for dynamic memory
    scaling based on the available system memory.
    """

    system_wide_memory_size: ByteSize | None
    """Total memory available in the whole system."""

    max_used_memory_ratio: float
    """The maximum acceptable ratio of `current_size` to `max_memory_size`."""

    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    """The time at which the system load information was measured."""

    @property
    def is_overloaded(self) -> bool:
        """Indicate whether the memory is considered as overloaded."""
        if self.system_wide_memory_size is not None and self.system_wide_used_size is not None:
            system_wide_utilization = self.system_wide_used_size / self.system_wide_memory_size
            if system_wide_utilization > SYSTEM_WIDE_MEMORY_OVERLOAD_THRESHOLD:
                return True

        if isinstance(self.max_memory_size, Ratio):
            # The snapshot overload is decided not when the snapshot was taken, but when `is_overload` property is
            # accessed. This allows for dynamic memory scaling. The same memory snapshot that used to be overloaded in
            # the past can become non-overloaded if the available memory was increased.
            max_memory_size = ByteSize(int(get_memory_info().total_size.bytes * self.max_memory_size.value))
        else:

Review comment (suggestion):
I believe the `available_memory_ratio` docstring in `Configuration` should also mention this dynamic scaling behavior.

Code after:
    used_ratio: float
    """The ratio of CPU currently in use."""

    max_used_ratio: float
    """The maximum ratio of CPU that is considered acceptable."""

    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    """The time at which the system load information was measured."""

    @property
    def is_overloaded(self) -> bool:
        """Indicate whether the CPU is considered as overloaded."""
        return self.used_ratio > self.max_used_ratio


@dataclass
class MemorySnapshot:
    """A snapshot of memory usage."""

    current_size: ByteSize
    """Memory usage of the current Python process and its children."""

    system_wide_used_size: ByteSize | None
    """Memory usage of all processes, system-wide."""

    max_memory_size: ByteSize
    """The maximum memory that can be used by `AutoscaledPool`."""

    system_wide_memory_size: ByteSize | None
    """Total memory available in the whole system."""

    max_used_memory_ratio: float
    """The maximum acceptable ratio of `current_size` to `max_memory_size`."""

    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    """The time at which the system load information was measured."""

    @property
    def is_overloaded(self) -> bool:
        """Indicate whether the memory is considered as overloaded."""
        if self.system_wide_memory_size is not None and self.system_wide_used_size is not None:
            system_wide_utilization = self.system_wide_used_size / self.system_wide_memory_size
            if system_wide_utilization > SYSTEM_WIDE_MEMORY_OVERLOAD_THRESHOLD:
                return True

        return (self.current_size / self.max_memory_size) > self.max_used_memory_ratio


@dataclass
class EventLoopSnapshot:
    """Snapshot of the state of the event loop."""