""" AUTHOR: Khushal P Soonderji DATE: CREATED: Tue, 25th Nov, 2025 UPDATED: Tue, 25th Nov, 2025 OBJECTIVE: To define a caching class that uses in-RAM dicts to hold cached data. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys from unittest import case sys.path.append(".") sys.path.append("..") # Other utils: from utils_v2.string import json from utils_v2.serialization.pickle_serializer import PickleSerializer from utils_v2.date_time import date_time from utils_v2.string import regex from utils_v2.cache_v2.base import AsyncCachingBase # To work with datatypes: from typing import List, Any # For async activities: import asyncio # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncLocalCache(AsyncCachingBase): def __init__( self, debug = False, debug_prefix = "R-Cache | " ): """ Implements a simple cache in RAM that holds and returns all native datatypes like ints, floats, bools, strings, dicts, lists, sets, and tuples :) :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 ) # Create a local dict that will hold the cached data: self.__cached_data = {} # ┳┓ ┓ • # ┃┃┏┓┣┓┓┏┏┓┏┓┓┏┓┏┓ # ┻┛┗ ┗┛┗┻┗┫┗┫┗┛┗┗┫ # ┛ ┛ ┛ def enable_debug(self): self.__printer.enable() def disable_debug(self): self.__printer.disable() # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ @staticmethod def now_utc() -> int | float: """ Returns the current UTC time as a timestamp. :return: The current UTC time as a timestamp. """ # Return the current time in UTC as a timestamp: return date_time.get_current_utc_date_time(as_string = False).timestamp() async def clear_expired_keys(self) -> None: """ Clears all the expired cached data. :return: None. """ # Keys to delete: to_del = [] # Enlist all the keys that need to be deleted: for k, v in self.__cached_data.items(): if self.now_utc() >= v["exp"]: to_del.append(k) # Delete the expired keys: for k in to_del: self.__cached_data.pop(k) # ┏┓ # ┃ ┏┓┏┓┏┓ # ┗┛┗┛┛ ┗ async def list_keys( self, match: str | None = 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 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. """ # Clear the expired keys: await self.clear_expired_keys() # Start with an empty list: keys_list = [] try: # Enlist all the keys, # test the pattern against all the keys and keep only those that match: for k in self.__cached_data.keys(): if match is None: keys_list.append(k) elif regex.match(text = str(k), pattern = match): keys_list.append(k) # 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. """ # Clear the expired keys: await self.clear_expired_keys() # Start by assuming failure: ttl = None try: # Get the expiry: exp_utc = self.__cached_data.get(key, {}).get("exp", None) if exp_utc is None: ttl = -2 else: ttl = exp_utc - self.now_utc() 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. """ # Clear the expired keys: await self.clear_expired_keys() try: # If the expiry is not specified, # make it an unreasonably far off future date: if expiry is None: expiry = date_time.get_current_utc_date_time(as_string = False) + date_time.timedelta(days = 365) expiry = expiry.timestamp() # Here we actually try to set the data: self.__cached_data[key] = { "val": value, "exp": self.now_utc() + expiry } # Done here: return True # 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. """ # Clear the expired keys: await self.clear_expired_keys() try: # Here we try to fetch the data: data = self.__cached_data.get(key, {}).get("val", None) 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. """ # Clear the expired keys: await self.clear_expired_keys() try: # Try to manually delete the key before expiry: response = self.__cached_data.pop(key, None) 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. """ # Clear the expired keys: await self.clear_expired_keys() try: # Check if already_existed = False if self.__cached_data.get(key, None) is None else True # If it already exists, we just increment the value; # else we create the value and increment the value starting from zero: if already_existed: self.__cached_data[key]["val"] = self.__cached_data[key]["val"] + value else: self.__cached_data[key] = { "val": value, "exp": self.now_utc() + expiry } # Return the new value: return self.__cached_data[key]["val"] # 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