450 lines
16 KiB
Python
450 lines
16 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
CREATED: Tue, 25th Nov, 2025
|
|
UPDATED: Tue, 25th Nov, 2025
|
|
|
|
OBJECTIVE:
|
|
|
|
To define a caching class that uses Redis to asynchronously cache information.
|
|
|
|
REFERENCES:
|
|
|
|
N/A
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
"""
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# To use redis:
|
|
import redis.asyncio as redis
|
|
from redis.asyncio.sentinel import Sentinel
|
|
|
|
# Other utils:
|
|
from utils_v2.string import json
|
|
from utils_v2.serialization.pickle_serializer import PickleSerializer
|
|
from utils_v2.cache_v2.base import AsyncCachingBase
|
|
|
|
# To work with datatypes:
|
|
from typing import List, Any
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** CLASSES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
class AsyncRedisCache(AsyncCachingBase):
|
|
|
|
def __init__(
|
|
self,
|
|
connection_string: str | dict = None,
|
|
serializer = None,
|
|
ping_counter = 1_000,
|
|
debug = False,
|
|
debug_prefix = "R-Cache | "
|
|
):
|
|
|
|
"""
|
|
Implements a simple cache in Redis which holds and returns all native datatypes like ints, floats, bools,
|
|
strings, dicts, lists, sets, and tuples :)
|
|
:param connection_string: The connection URL or Sentinel JSON for connecting to Redis.
|
|
:param serializer: The serializer to use.
|
|
:param ping_counter: The number of requests to Redis after which you want to ping to ensure connection.
|
|
:param debug: Whether, or not, you want to show debugging messages from the start.
|
|
:param debug_prefix: The prefix text to show with the debugging messages.
|
|
"""
|
|
|
|
# Invoke the parent class's constructor:
|
|
super().__init__(
|
|
debug = debug,
|
|
debug_prefix = debug_prefix
|
|
)
|
|
|
|
# Note down the configuration:
|
|
self.__client = None
|
|
self.__serializer = serializer or PickleSerializer()
|
|
self.__ping_counter = ping_counter
|
|
self.__requests_since_last_ping = 0
|
|
|
|
# Figure out the connection mechanism:
|
|
if isinstance(connection_string, dict):
|
|
self.__sentinel_json = connection_string
|
|
self.__connection_string = None
|
|
self.__printer("Received Sentinel JSON.")
|
|
elif isinstance(connection_string, str):
|
|
try:
|
|
self.__sentinel_json = json.from_string(connection_string)
|
|
self.__connection_string = None
|
|
self.__printer("Seems like a Sentinel JSON String.")
|
|
except:
|
|
self.__sentinel_json = None
|
|
self.__connection_string = connection_string
|
|
self.__printer("Seems like a regular Connection String.")
|
|
|
|
# ┳┓ ┓ •
|
|
# ┃┃┏┓┣┓┓┏┏┓┏┓┓┏┓┏┓
|
|
# ┻┛┗ ┗┛┗┻┗┫┗┫┗┛┗┗┫
|
|
# ┛ ┛ ┛
|
|
|
|
def enable_debug(self):
|
|
self.__printer.enable()
|
|
|
|
def disable_debug(self):
|
|
self.__printer.disable()
|
|
|
|
# ┏┓ •
|
|
# ┃ ┏┓┏┓┏┓┏┓┏╋┓┏┓┏┓
|
|
# ┗┛┗┛┛┗┛┗┗ ┗┗┗┗┛┛┗
|
|
|
|
async def connect(self) -> bool:
|
|
|
|
"""
|
|
To make an asynchronous connection request to the Redis server to establish a connection.
|
|
:return: True if connected, else False.
|
|
"""
|
|
|
|
try:
|
|
|
|
# First priority given to the cluster connection if a sentinel JSON is provided.
|
|
# This provides robustness against the failure of one single node.
|
|
if self.__sentinel_json is not None:
|
|
sentinel_conn = Sentinel(
|
|
sentinels = [(s["host"], s["port"]) for s in self.__sentinel_json["servers"]],
|
|
socket_timeout = 2.5
|
|
)
|
|
self.__client = sentinel_conn.master_for(
|
|
service_name = self.__sentinel_json["serviceName"],
|
|
db = self.__sentinel_json["db"],
|
|
password = self.__sentinel_json["password"]
|
|
)
|
|
return True
|
|
|
|
# If a cluster setup isn't ready,
|
|
# we connect to a single node through a simple connection string:
|
|
elif self.__connection_string is not None:
|
|
self.__client = redis.from_url(
|
|
self.__connection_string
|
|
)
|
|
return True
|
|
|
|
# When none of the available connection mechanisms are given:
|
|
else: raise ValueError("Either a Sentinel JSON or a Connection String is needed.")
|
|
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
return False
|
|
|
|
async def disconnect(self) -> bool:
|
|
|
|
"""
|
|
Close the connection to the Redis server.
|
|
:return: True if successful, else False.
|
|
"""
|
|
|
|
if self.__client is not None:
|
|
try: await self.__client.close()
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
return False
|
|
return True
|
|
return True
|
|
|
|
async def ensure_connection(self) -> bool:
|
|
|
|
"""
|
|
To ensure that we are connected. We keep pinging the Redis server every once in a while even when connected.
|
|
:return: True if connected, else False.
|
|
"""
|
|
|
|
# If we are not connected, we try to establish a connection:
|
|
if self.__client is None: return await self.connect()
|
|
|
|
# Else we check if we are connected. If not, we try to connect.
|
|
# But we check only once in a while. In the meantime, we assume that we are connected.
|
|
elif self.__requests_since_last_ping > self.__ping_counter:
|
|
try:
|
|
await self.__client.ping()
|
|
self.__requests_since_last_ping = 0
|
|
return True
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
return await self.connect()
|
|
else:
|
|
self.__requests_since_last_ping += 1
|
|
return True
|
|
|
|
# ┏┓
|
|
# ┃ ┏┓┏┓┏┓
|
|
# ┗┛┗┛┛ ┗
|
|
|
|
async def list_keys(
|
|
self,
|
|
match: str = "*",
|
|
raise_exception: bool = False
|
|
) -> List[str] | None:
|
|
|
|
"""
|
|
To get a list of all the keys that match the pattern.
|
|
:param match: To match a glob-style pattern. THIS IS NOT FULL-FLEDGED REGEX.
|
|
:param raise_exception: Whether to raise or suppress exceptions.
|
|
:return: The list of keys if successful, else None.
|
|
"""
|
|
|
|
# Standard connectivity check:
|
|
await self.ensure_connection()
|
|
|
|
# Start with an empty list:
|
|
keys_list = []
|
|
|
|
try:
|
|
|
|
# Start at the beginning,
|
|
# and break out of the loop if the pointer returns to zero:
|
|
pointer = 0
|
|
while True:
|
|
pointer, keys = await self.__client.scan(pointer, match = match)
|
|
keys_list = keys_list + [k.decode("utf-8") for k in keys]
|
|
if pointer == 0: break
|
|
|
|
# Done here:
|
|
return list(set(keys_list)) if keys_list else None
|
|
|
|
# If an exception occurs in th eprocess:
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
if raise_exception: raise
|
|
else: return None
|
|
|
|
async def ttl(
|
|
self,
|
|
key: str | bytes,
|
|
raise_exception: bool = False
|
|
) -> int | float | None:
|
|
|
|
"""
|
|
To get the no. of seconds till the expiry of some key.
|
|
:param key: The key to check the expiry of.
|
|
:param raise_exception: Whether to raise or suppress exceptions.
|
|
:return: -1 if the key is persistent (i.e., no expiry time set), -2 if the key does not exist, or the time left
|
|
in seconds if the key exists and is not persistent. If something goes wrong, you will get a null value.
|
|
"""
|
|
|
|
# Standard connectivity check:
|
|
await self.ensure_connection()
|
|
|
|
# Start by assuming failure:
|
|
ttl = None
|
|
|
|
try:
|
|
|
|
# Get the TTL of the key in ms,
|
|
# convert it to seconds and return the value:
|
|
ttl_ms = await self.__client.pttl(key)
|
|
if ttl_ms >= 0: ttl = ttl_ms / 1_000
|
|
return ttl
|
|
|
|
# If an exception occurs in th eprocess:
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
if raise_exception: raise
|
|
else: return None
|
|
|
|
async def set(
|
|
self,
|
|
key: str | bytes,
|
|
value: Any,
|
|
expiry: float = None,
|
|
raise_exception: bool = False
|
|
) -> bool:
|
|
|
|
"""
|
|
Saves some value to the cache. If an expiry is specified, the data will be deleted after that many seconds.
|
|
:param key: The key with which the data will be stored and retrieved.
|
|
:param value: The value to store.
|
|
:param expiry: The time in seconds after which the data will expire. Must be a positive number.
|
|
:param raise_exception: Whether to raise or suppress exceptions.
|
|
:return: True if cached, else False.
|
|
"""
|
|
|
|
# Standard connectivity check:
|
|
await self.ensure_connection()
|
|
|
|
try:
|
|
|
|
# Here we actually try to set the data:
|
|
data = self.__serializer.serialize(value)
|
|
if expiry: response = await self.__client.setex(key, int(expiry), data)
|
|
else: response = await self.__client.set(key, data)
|
|
return response
|
|
|
|
# If an exception occurs in th eprocess:
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
if raise_exception: raise
|
|
else: return False
|
|
|
|
async def get(
|
|
self,
|
|
key: str | bytes,
|
|
raise_exception: bool = False,
|
|
on_fail: Any = None
|
|
) -> Any:
|
|
|
|
"""
|
|
Retrieve the cached value.
|
|
:param key: The key with which the data was saved.
|
|
:param raise_exception: Whether to raise or suppress exceptions.
|
|
:param on_fail: What to return if the process fails due to an exception.
|
|
:return: The retrieved data or null if not found.
|
|
"""
|
|
|
|
# Standard connectivity check:
|
|
await self.ensure_connection()
|
|
|
|
try:
|
|
|
|
# Here we try to fetch the data:
|
|
data = await self.__client.get(key)
|
|
data = self.__serializer.deserialize(data)
|
|
return data
|
|
|
|
# If an exception occurs in th eprocess:
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
if raise_exception: raise
|
|
else: return on_fail
|
|
|
|
async def delete(
|
|
self,
|
|
key: str | bytes,
|
|
raise_exception: bool = False
|
|
) -> bool:
|
|
|
|
"""
|
|
Prematurely delete the value from the cache before it expires.
|
|
:param key: The key with which the data was saved.
|
|
:param raise_exception: Whether to raise or suppress exceptions.
|
|
:return: True if deleted, else False.
|
|
"""
|
|
|
|
# Standard connectivity check:
|
|
await self.ensure_connection()
|
|
|
|
try:
|
|
|
|
# Try to manually delete the key before expiry:
|
|
response = await self.__client.delete(key)
|
|
return True if response else False
|
|
|
|
# If an exception occurs in th eprocess:
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
if raise_exception: raise
|
|
else: return False
|
|
|
|
async def count(
|
|
self,
|
|
key: str | bytes,
|
|
value: int = 1,
|
|
expiry: float = None,
|
|
raise_exception: bool = False
|
|
) -> int | None:
|
|
|
|
"""
|
|
To use simple counters. If the counter (identified by the 'key') exists, it will be incremented, else the
|
|
counter will be created and the value will be incremented from 0.
|
|
:param key: The name of the counter.
|
|
:param value: The amount to increment the value by. Send negative values to count backwards.
|
|
:param expiry: The time (in seconds) in which the counter expires. Starts from the time the counter is created.
|
|
This value has to be an integer. If a float is passed, the value will be rounded off.
|
|
:param raise_exception: Whether to raise or suppress exceptions.
|
|
:return: The latest value of the counter. Will be null if something went wrong and the exception was suppressed.
|
|
"""
|
|
|
|
# Standard connectivity check:
|
|
await self.ensure_connection()
|
|
|
|
try:
|
|
|
|
# check if the key already exists,
|
|
# regardless of that, increment the counter:
|
|
already_existed = await self.__client.exists(key)
|
|
new_value = await self.__client.incrby(key, value)
|
|
|
|
# If the key didn't already exist, specify the expiry:
|
|
if expiry and not already_existed: await self.__client.expire(key, int(expiry))
|
|
|
|
# Done here:
|
|
return new_value
|
|
|
|
# If an exception occurs in th eprocess:
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
if raise_exception: raise
|
|
else: return None
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
pass
|