Method keys

cache/redis_cache.py:77–98

Get all keys matching the pattern First try KEYS command, if not supported fallback to SCAN

(self, pattern: str)

        self._redis_client.set(key, pickle.dumps(value), ex=expire_time)

    def keys(self, pattern: str) -> List[str]:
        """
        Get all keys matching the pattern
        First try KEYS command, if not supported fallback to SCAN
        """
        try:
            # Try KEYS command first (faster for standard Redis)
            return [key.decode() if isinstance(key, bytes) else key for key in self._redis_client.keys(pattern)]
        except ResponseError as e:
            # If KEYS is not supported (e.g., Redis Cluster or cloud Redis), use SCAN
            if "unknown command" in str(e).lower() or "keys" in str(e).lower():
                keys_list: List[str] = []
                cursor = 0
                while True:
                    cursor, keys = self._redis_client.scan(cursor=cursor, match=pattern, count=100)
                    keys_list.extend([key.decode() if isinstance(key, bytes) else key for key in keys])
                    if cursor == 0:
                        break
                return keys_list
            else:
                # Re-raise if it's a different error
                raise


if __name__ == '__main__':