(20251125) added a new kind of caching that is similar to the Async Redis Cache module that we had previously, but now allows more kinds of cache to be used.
This commit is contained in:
@@ -0,0 +1,390 @@
|
||||
"""
|
||||
|
||||
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 = ".*",
|
||||
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 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
|
||||
@@ -0,0 +1,449 @@
|
||||
"""
|
||||
|
||||
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
|
||||
@@ -0,0 +1,361 @@
|
||||
"""
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user