""" # Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License" # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ from abc import ABC, abstractmethod from typing import Any, List, Optional import paddle from fastdeploy.utils import get_logger logger = get_logger("cache_storage", "cache_storage.log") class KVCacheStorage(ABC): """ KVCacheStorage is a class that provides a generic key-value interface for storing and retrieving KV cache. """ @abstractmethod def get( self, key: str, target_location: Optional[Any] = None, target_size: Optional[Any] = None, ) -> paddle.Tensor | None: """ Retrieve the value associated with the given key. Returns None if the key does not exist. """ pass @abstractmethod def batch_get( self, keys: List[str], target_locations: Optional[Any] = None, target_sizes: Optional[Any] = None, ) -> List[paddle.Tensor | None]: """ Retrieve values for multiple keys. Returns a list of tensors or None for each key. """ pass @abstractmethod def set( self, key: str, target_location: Optional[Any] = None, target_size: Optional[Any] = None, ) -> bool: """ Store the value associated with the given key. Returns True if the operation was successful, False otherwise. """ pass @abstractmethod def batch_set( self, keys: List[str], target_locations: Optional[Any] = None, target_sizes: Optional[Any] = None, ) -> bool: """ Store multiple key-value pairs. Returns True if all operations were successful, False otherwise. """ pass @abstractmethod def exists(self, keys: List[str]) -> bool: """ Check if the key exists in the storage. Returns True if the key exists, False otherwise. """ pass @abstractmethod def clear(self) -> bool: """ Clear all keys in storage """ pass