File size: 3,284 Bytes
c446951 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 |
from contextlib import contextmanager
from typing import Any, Optional
class BaseCache:
"""
BaseCache is an abstract base class that defines the interface for a cache.
"""
def get(self, key: str):
"""
Gets the value associated with the given key.
Args:
key (str): The key to retrieve the value.
Raises:
NotImplementedError: This method must be implemented by subclasses.
"""
raise NotImplementedError()
def set(self, key: str, value: str, expire: float = None):
"""
Sets a value for a given key with an optional expire time.
Args:
key (str): The key to store the value.
value (str): The value to store.
expire (float, optional): The time, in seconds, after which the key will expire. Defaults to None.
Raises:
NotImplementedError: This method must be implemented by subclasses.
"""
raise NotImplementedError()
def zadd(self, key: str, value: str, score: float, expire: float = None):
"""
Adds a member with the specified score to the sorted set stored at key.
Args:
key (str): The key of the sorted set.
value (str): The value to add to the sorted set.
score (float): The score associated with the value.
expire (float, optional): The time, in seconds, after which the key will expire. Defaults to None.
Raises:
NotImplementedError: This method must be implemented by subclasses.
"""
raise NotImplementedError()
def zrangebyscore(
self,
key: str,
min: Optional[float] = -1,
max: Optional[float] = float("inf"),
withscores: bool = False,
):
"""
Retrieves a range of members from a sorted set.
Args:
key (str): The key of the sorted set.
start (int, optional): The starting index of the range. Defaults to -1.
stop (int, optional): The ending index of the range. Defaults to float("inf").
withscores (bool, optional): Whether to return the scores along with the values. Defaults to False.
Raises:
NotImplementedError: This method must be implemented by subclasses.
"""
raise NotImplementedError()
def zremrangebyscore(
self,
key: str,
start: Optional[int] = -1,
stop: Optional[int] = float("inf"),
):
"""
Removes all members in a sorted set within the given scores.
Args:
key (str): The key of the sorted set.
start (int, optional): The minimum score of the range. Defaults to -1.
stop (int, optional): The maximum score of the range. Defaults to float("inf").
Raises:
NotImplementedError: This method must be implemented by subclasses.
"""
raise NotImplementedError()
def acquire_lock(self, key: str, expire: float = None) -> Any:
raise NotImplementedError()
@contextmanager
def lock(self, key: str, expire: float = None) -> Any:
l = self.acquire_lock(key, expire=expire)
try:
yield l
finally:
l.release()
|