""" AUTHOR: Khushal P Soonderji DATE: CREATED: Tue, 25th Nov, 2025 UPDATED: Tue, 25th Nov, 2025 OBJECTIVE: To define a set of functions for caching requirements. We will have caching decorators as well as an abstract class that defines the required methods. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For defining abstract classes: from abc import ABC, abstractmethod # For debugging: from icecream import IceCreamDebugger # For hashing and shortening the hash: import hashlib import base64 # Other utils: from utils_v2.string import json # To work with datatypes: from typing import List, Any # To make decorators: from functools import wraps # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncCachingBase(ABC): def __init__( self, debug: bool = True, debug_prefix: str = "A-Cache | " ): """ The constructor of the caching abstract class :param debug: Whether, or not, you would like to print debugging messages. :param debug_prefix: The prefix text to show with the debugging messages. """ # For debugging: self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) if not debug: self.__printer.disable() # ┳┓ ┓ • # ┃┃┏┓┣┓┓┏┏┓┏┓┓┏┓┏┓ # ┻┛┗ ┗┛┗┻┗┫┗┫┗┛┗┗┫ # ┛ ┛ ┛ def enable_debug(self): self.__printer.enable() def disable_debug(self): self.__printer.disable() # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ @staticmethod def make_key(*args, **kwargs) -> str: """ Generates a key by hashing the args and kwargs sent to it. Can be useful to generate a predictable key. If the inputs stay the same, the output stays the same. :param args: Any number of args that you would like to use to generate the key. :param kwargs: Any number of kwargs that you would like to use to generate the key. :return: A string that can be used as a key to store values on Redis. """ # Take everything into one plain text string: plain_text = "".join(str(a) for a in args) plain_text += json.to_string(kwargs, no_space = True) # Hash the plain text value, and create a key from it: sha256_hash = hashlib.sha256() sha256_hash.update(plain_text.encode("utf-8")) hashed_key = sha256_hash.digest() base64_key = base64.b64encode(hashed_key).decode("utf-8") # Done here: return base64_key # ┏┓ # ┃ ┏┓┏┓┏┓ # ┗┛┗┛┛ ┗ @abstractmethod async def list_keys( self, match: str = None, raise_exception: bool = False ) -> List[str] | None: """ To get a list of all the keys that match the pattern. :param match: To match a RegEx pattern. :param raise_exception: Whether to raise or suppress exceptions. :return: The list of keys if successful, else None. """ raise NotImplementedError @abstractmethod 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. """ raise NotImplementedError @abstractmethod 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. """ raise NotImplementedError @abstractmethod 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 or 'on_fail' if not found. """ raise NotImplementedError @abstractmethod 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. """ raise NotImplementedError @abstractmethod 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. """ raise NotImplementedError # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** def cache_it( cache: AsyncCachingBase, expiry: int | float = 120.0 ): """ This decorator factory takes an instance of the async caching class 'AsyncCachingBase' and holds your data there. If a subsequent call is made to the same decorated function with the same inputs, the result is fetched from the cache instead of going through the whole function again. :param cache: The instance of 'AsyncCachingBase'. :param expiry: The time in seconds after which the cached data must be cleared. :return: The decorator that automatically caches your data. """ def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): # We first use the name of the function and the inputs given to it to generate a key for caching with the # simple hashing and shortening by way of base64 strings: inputs_given = func.__name__ + str([_ for _ in args]) + str(kwargs) base64_key = cache.make_key(inputs_given) # Now we check if we have the value in cache: try: response = await cache.get(base64_key, raise_exception = True) # If the key doesn't exist, we pass through the function and store the results. except: response = await func(*args, **kwargs) await cache.set(key = base64_key, value = response, expiry = expiry) # Return the response from the wrapped function. return response return wrapper return decorator # --------------------------------------------------------------------------------------------------------------------- def cache_class_methods( attr_name: str, expiry: int | float = 120.0 ): """ This decorator factory takes an instance of the async caching class 'AsyncCachingBase' and holds your data there. If a subsequent call is made to the same decorated function with the same inputs, the result is fetched from the cache instead of going through the whole function again. :param attr_name: The name of the variable that has an instance of 'AsyncCachingBase'. :param expiry: The time in seconds after which the cached data must be cleared. :return: The decorator that automatically caches your data. """ def decorator(func): @wraps(func) async def wrapper(self, *args, **kwargs): # Get the cache object first: cache_obj = getattr(self, attr_name) # We first use the name of the function and the inputs given to it to generate a key for caching with the # simple hashing and shortening by way of base64 strings: inputs_given = func.__name__ + str([_ for _ in args]) + str(kwargs) base64_key = cache_obj.make_key(inputs_given) # Now we check if we have the value in cache: try: response = await cache_obj.get(base64_key, raise_exception = True) # If the key doesn't exist, we pass through the function and store the results. except: response = await func(self, *args, **kwargs) await cache_obj.set(key = base64_key, value = response, expiry = expiry) # Return the response from the wrapped function. return response return wrapper return decorator # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass