""" 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. This is truly stateless caching since yu can run your backend script from any number of servers and yet have the same cache data sync'd through Redis. 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.async_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_params: 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_params: 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_params, dict): self.__sentinel_json = connection_params self.__connection_string = None self._printer("Received Sentinel JSON.") elif isinstance(connection_params, str): try: self.__sentinel_json = json.from_string(connection_params) self.__connection_string = None self._printer("Seems like a Sentinel JSON String.") except: self.__sentinel_json = None self.__connection_string = connection_params self._printer("Seems like a regular Connection String.") # ┏┓ • # ┃ ┏┓┏┓┏┓┏┓┏╋┓┏┓┏┓ # ┗┛┗┛┛┗┛┗┗ ┗┗┗┗┛┛┗ 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("CONN. ERR!", 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("LIST ERR!", 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("TTL ERR!", key, 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("SET ERR!", key, 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("CACHE MISS!", key, 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("DEL ERR!", key, 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("COUNT ERR!", key, value, exception) if raise_exception: raise else: return None # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass