From a10120f45bb43d7c2bf6eb8dd15af9cbcc51d095 Mon Sep 17 00:00:00 2001 From: khushal Date: Tue, 26 Nov 2024 11:45:21 +0530 Subject: [PATCH] (20241126) Upgraded Redis Cache v2. It can now list keys and get the TTL for existing keys. --- utils_v2/cache/async_redis_cache_v2.py | 133 +++++++++++++++++++++++-- utils_v2/goog/gmail.py | 55 +++++++++- 2 files changed, 179 insertions(+), 9 deletions(-) diff --git a/utils_v2/cache/async_redis_cache_v2.py b/utils_v2/cache/async_redis_cache_v2.py index bda0780..0192dae 100644 --- a/utils_v2/cache/async_redis_cache_v2.py +++ b/utils_v2/cache/async_redis_cache_v2.py @@ -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") diff --git a/utils_v2/goog/gmail.py b/utils_v2/goog/gmail.py index 00ccd23..437f191 100644 --- a/utils_v2/goog/gmail.py +++ b/utils_v2/goog/gmail.py @@ -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 + # ***************************************************************************************************************** # ***** ****