Repository: apify/crawlee-python
File: src/crawlee/storages/_key_value_store.py (Python)

Code before:
            The public URL for the given key.
        """
        return await self._resource_client.get_public_url(key)

    async def get_auto_saved_value(self, key: str, default_value: dict[str, Any] | None = None) -> dict[str, Any]:
        """Gets a value from store that will be automatically saved on changes.

        Args:
            key: Key of the record, to store the value.
            default_value: value to be used if the record does not exist yet. Should be a dictionary

        Returns:
            Returns the value of the key
        """
        default_value = {} if default_value is None else default_value

        if key in self._cache:
            return self._cache[key]

        value = await self.get_value(key, default_value)

        self._cache[key] = value

        self._ensure_persist_event()

        return cast(dict[str, Any], value)

    async def _persist_save(self, _event_data: EventPersistStateData | None = None) -> None:
        """Save cache with persistent values. Can be used in Event Manager."""
        for key, value in self._cache.items():
            await self.set_value(key, value)

    def _ensure_persist_event(self) -> None:
        """Setup persist state event handling if not already done."""
        if self._persist_state_event_started:
            return

        event_manager = service_container.get_event_manager()
        event_manager.on(event=Event.PERSIST_STATE, listener=self._persist_save)
        self._persist_state_event_started = True

    def clear_cache(self) -> None:
        """Clear cache with persistent values."""
        self._cache.clear()

    def drop_persist_state_event(self) -> None:
        """Off event_manager listener and drop event status."""
        if self._persist_state_event_started:
            event_manager = service_container.get_event_manager()
            event_manager.off(event=Event.PERSIST_STATE, listener=self._persist_save)
        self._persist_state_event_started = False

Review comment (suggestion):
I believe the cast is not necessary, correct?

The returned value of `get_value` is the same type of `default_value`, which is `dict[str, Any]`.

Code after:
        Returns:
            The public URL for the given key.
        """
        return await self._resource_client.get_public_url(key)

    async def get_auto_saved_value(
        self, key: str, default_value: dict[str, JsonSerializable] | None = None
    ) -> dict[str, JsonSerializable]:
        """Gets a value from KVS that will be automatically saved on changes.

        Args:
            key: Key of the record, to store the value.
            default_value: Value to be used if the record does not exist yet. Should be a dictionary.

        Returns:
            Returns the value of the key.
        """
        default_value = {} if default_value is None else default_value

        if key in self._cache:
            return self._cache[key]

        value = await self.get_value(key, default_value)

        if not isinstance(value, dict):
            raise TypeError(
                f'Expected dictionary for persist state value at key "{key}, but got {type(value).__name__}'
            )

        self._cache[key] = value

        self._ensure_persist_event()

        return value

    @property
    def _cache(self) -> dict[str, dict[str, JsonSerializable]]:
        """Cache dictionary for storing auto-saved values indexed by store ID."""
        if self._id not in self._general_cache:
            self._general_cache[self._id] = {}
        return self._general_cache[self._id]

    async def _persist_save(self, _event_data: EventPersistStateData | None = None) -> None:
        """Save cache with persistent values. Can be used in Event Manager."""
        for key, value in self._cache.items():
            await self.set_value(key, value)

    def _ensure_persist_event(self) -> None:
        """Setup persist state event handling if not already done."""
        if self._persist_state_event_started:
            return