khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241126) Upgraded Redis Cache v2. It can now list keys and get the TTL for existing keys.

Browse Source
This commit is contained in:
khushal
2024-11-26 11:45:21 +05:30
parent 87593baac5
commit a10120f45b
2 changed files with 179 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/cache/async_redis_cache_v2.py
Vendored
+125 -8
View File
@@ -47,6 +47,9 @@ from functools import wraps
import hashlib
import base64
# To work with datatypes:
from typing import List, Any
# *****************************************************************************************************************
# ***** ****
@@ -217,7 +220,7 @@ class AsyncRedisCache:
self.__printer.disable()
@staticmethod
def make_key(*args, **kwargs):
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
@@ -240,7 +243,7 @@ class AsyncRedisCache:
# Done here:
return base64_key
async def connect(self):
async def connect(self) -> bool:
"""
To make an asynchronous connection request to the Redis server to establish a connection.
@@ -257,7 +260,7 @@ class AsyncRedisCache:
self.__printer(exception)
return False
async def disconnect(self):
async def disconnect(self) -> bool:
"""
Close the connection to the Redis server.
@@ -272,7 +275,7 @@ class AsyncRedisCache:
return True
return True
async def ensure_connection(self):
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.
@@ -296,7 +299,85 @@ class AsyncRedisCache:
self.__requests_since_last_ping += 1
return True
async def set(self, key, value, expiry: float = None, raise_exception = False):
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: If you want to raise an exception if the process fails.
: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:
: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.
@@ -307,6 +388,7 @@ class AsyncRedisCache:
:return: True if cached, else False.
"""
# Standard connectivity check:
await self.ensure_connection()
try:
@@ -317,12 +399,18 @@ class AsyncRedisCache:
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, raise_exception = False, on_fail = None):
async def get(
self,
key: str | bytes,
raise_exception: bool = False,
on_fail: Any = None
) -> Any:
"""
Retrieve the cached value.
@@ -332,6 +420,7 @@ class AsyncRedisCache:
:return: The retrieved data or null if not found.
"""
# Standard connectivity check:
await self.ensure_connection()
try:
@@ -341,12 +430,17 @@ class AsyncRedisCache:
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, raise_exception = False):
async def delete(
self,
key: str | bytes,
raise_exception: bool = False
) -> bool:
"""
Prematurely delete the value from the cache before it expires.
@@ -355,6 +449,7 @@ class AsyncRedisCache:
:return: True if deleted, else False.
"""
# Standard connectivity check:
await self.ensure_connection()
try:
@@ -363,12 +458,19 @@ class AsyncRedisCache:
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, value: int = 1, expiry: float = None, raise_exception = 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
@@ -381,15 +483,23 @@ class AsyncRedisCache:
: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
@@ -422,6 +532,13 @@ if __name__ == "__main__":
# Proceed only if we connected to the database successfully:
if connected:
existing_keys = await my_cache.list_keys()
print(f"KEYS ({len(existing_keys)}):", json.to_string(existing_keys, default = str))
key = str(input("Paste a key to check its TTL: "))
key_ttl = await my_cache.ttl(key = key)
print(f"KEY '{key}' has {key_ttl} seconds of TTL.")
value = await my_cache.get(key = str(input("Paste a key to get: ")))
print("GET:", json.to_string(value, default = str) if isinstance(value, (dict, list)) else value)
print("TYP:", type(value), end = "\n\n")
utils_v2/goog/gmail.py
+54 -1
View File
@@ -60,7 +60,7 @@ import asyncio
import datetime
# For working with datatypes:
from typing import Dict, Literal
from typing import Dict, Literal, List
# For debugging:
from icecream import IceCreamDebugger
@@ -502,6 +502,59 @@ class AsyncGMailClient:
# Done here:
return success
# ┳┳┓
# ┃┃┃┏┓┏┏┏┓┏┓┏┓┏
# ┛ ┗┗ ┛┛┗┻┗┫┗ ┛
# ┛
async def __list_messages_in_page(
self,
count: int = 100,
query: str = None,
label_ids: List[str] | str = None,
include_spam_and_trash: bool = False,
next_page_token: str = None,
raise_exception: bool = False
):
# Start by assuming failure:
page_messages = None
try:
# Standard token-refresh check:
await self.__ensure_token()
# Build the needed params:
params_json = {"count": count}
if query: params_json["q"] = query
if next_page_token: params_json["pageToken"] = next_page_token
if label_ids: params_json["labelIds"] = label_ids if isinstance(label_ids, list) else [label_ids]
if include_spam_and_trash: params_json["includeSpamTrash"] = include_spam_and_trash
# Make the API call:
if not self._debug_only_errors: self._printer("Listing All Labels.", self.__user_email)
api_response = await self.__http_client.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/messages",
headers = {"Authorization": f"Bearer {self.__credentials.token}"},
params = params_json
)
# If the API call failed:
if api_response.status_code not in [200]: return labels
# Else we format the response:
labels = {label.pop("name"): label for label in api_response.json().get("labels", [])}
# In case something goes wrong along the way:
except Exception as exception:
if raise_exception: raise
self._printer(exception)
labels = None
# Done here:
return labels
# *****************************************************************************************************************
# ***** ****