From ed415c6b3a9a2548ebc5a168db3220a9d0aed69e Mon Sep 17 00:00:00 2001 From: khushal Date: Mon, 25 Nov 2024 19:21:33 +0530 Subject: [PATCH] (20241125) GMail Client Started. Label-Management Ready. --- utils_v2/goog/__init__.py | 0 utils_v2/goog/gmail.py | 582 ++++++++++++++++++++++++++++++++ utils_v2/oauth/services/goog.py | 59 ++-- 3 files changed, 622 insertions(+), 19 deletions(-) create mode 100644 utils_v2/goog/__init__.py create mode 100644 utils_v2/goog/gmail.py diff --git a/utils_v2/goog/__init__.py b/utils_v2/goog/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/goog/gmail.py b/utils_v2/goog/gmail.py new file mode 100644 index 0000000..9410f9d --- /dev/null +++ b/utils_v2/goog/gmail.py @@ -0,0 +1,582 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Monday, 25th Nov., 2024 + + OBJECTIVE: + + To manage e-mails in a GMail account. + + REFERENCES: + + 1. Quickstart: https://developers.google.com/gmail/api/quickstart/python + 2. Labels: https://developers.google.com/gmail/api/guides/labels + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# System-level activities: +import io + +# My utils: +from utils_v2.string import json +from utils_v2.date_time import date_time +from utils_v2.oauth.services.goog import GoogleOAuth + +# Related to Google: +from google.auth.transport.requests import Request +from google.oauth2.credentials import Credentials +from googleapiclient.discovery import build + +# To make API calls: +import httpx + +# For asynchronous activities: +import asyncio + +# To work with date and time: +import datetime + +# For working with datatypes: +from typing import Dict, Literal + +# For debugging: +from icecream import IceCreamDebugger + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class AsyncGMailClient: + + def __init__( + self, + credentials: Credentials, + http_client: httpx.AsyncClient, + debug = True, + debug_prefix = "GMail | ", + debug_only_errors = True + ): + + # Prepare the debugging utility: + self._debug_prefix = debug_prefix + self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) + if not debug: self._printer.disable() + self._debug_only_errors = debug_only_errors + + # Accept the input configuration: + self.__http_client = http_client + self.__service = None + self.__user_email = None + self.__credentials = credentials + + def enable_debug(self): + self._printer.enable() + + def disable_debug(self): + self._printer.disable() + + def debug_only_errors(self): + self._debug_only_errors = True + + def debug_everything(self): + self._debug_only_errors = False + + # ┏┓ ┏┓ ┏┓ ┏• + # ┗┓┏┓╋┓┏┏┓ ┣╋ ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗ ┗┗┻┣┛ ┗┻ ┗┛┗┛┛┗┛┗┗┫ + # ┛ ┛ + + async def initialize(self): + + """ + Call this once when the instance is created. + :return: True if successful, else False. + """ + + return await self.__build_service() + + async def __build_service(self) -> bool: + + """ + Build a service object that can be used to perform various activities. + :return: True if successful, else False. + """ + + try: + + # Build the service object: + self.__service = build( + serviceName = "gmail", + version = "v1", + credentials = self.__credentials, + cache_discovery = False + ) + + # Note down the e-mail address of the user: + profile = self.__service.users().getProfile(userId = "me").execute() + self.__user_email = profile.get("emailAddress") + + return True + + # In case something goes wrong: + except Exception as exception: + self._printer(exception) + return False + + async def set_credentials( + self, + credentials: Credentials + ) -> bool: + + """ + To update the credentials. Needed for the times when the access token gets refreshed. + :param credentials: Google's custom 'Credentials' object that describes the OAuth-based access details. + :return: True if successful, else False. + """ + + self.__credentials = credentials + return await self.__build_service() + + async def __ensure_token(self) -> None: + + """ + Checks if the access token has expired and refreshes if needed. + :return: None. + """ + + if await GoogleOAuth.credentials_have_expired(self.__credentials): + self._printer("Refreshing Token:", self.__user_email) + self.__credentials.refresh(Request()) + + # ┏┓ • + # ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏ + # ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛ + # ┛ + + @property + async def user_email(self): + return self.__user_email + + @property + async def user_profile(self): + return self.__service.users().getProfile(userId = "me").execute() + + # ┓ ┓ ┓ + # ┃ ┏┓┣┓┏┓┃┏ + # ┗┛┗┻┗┛┗ ┗┛ + + async def list_labels( + self, + raise_exception: bool = False + ) -> Dict[str, dict] | None: + + """ + To get the list of labels of this user. + DOCUMENTATION: + 1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/list + 2. https://developers.google.com/gmail/api/reference/rest/v1/users.labels#Label + :param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be + suppressed. + :return: The list of labels. + """ + + # Start by assuming failure: + labels = None + + try: + + # Standard token-refresh check: + await self.__ensure_token() + + # Make the API call: + api_response = await self.__http_client.get( + url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/labels", + headers = {"Authorization": f"Bearer {self.__credentials.token}"} + ) + + # 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 + + async def get_label( + self, + label_id: str, + raise_exception: bool = False + ) -> dict | None: + + """ + To get one label of this user. the label will be identified by its id. + DOCUMENTATION: + 1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/get + 2. https://developers.google.com/gmail/api/reference/rest/v1/users.labels#Label + :param label_id: The id that Google assigned to the label. + :param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be + suppressed. + :return: The list of labels. + """ + + # Start by assuming failure: + label = None + + try: + + # Standard token-refresh check: + await self.__ensure_token() + + # Make the API call: + api_response = await self.__http_client.get( + url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/labels/{label_id}", + headers = {"Authorization": f"Bearer {self.__credentials.token}"} + ) + + # If the API call failed: + if api_response.status_code not in [200]: return label + + # Else we extract the response: + label = api_response.json() + + # In case something goes wrong along the way: + except Exception as exception: + if raise_exception: raise + self._printer(exception) + label = None + + # Done here: + return label + + async def create_label( + self, + label_name: str, + label_visibility: Literal["labelShow", "labelShowIfUnread", "labelHide"] = "labelShow", + message_visibility: Literal["show", "hide"] = "show", + label_text_color: str = "#434343", + label_background_color: str = "#000000", + raise_exception: bool = False + ) -> bool: + + """ + Create one label for the user. + DOCUMENTATION: + 1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/create + 2. https://developers.google.com/gmail/api/reference/rest/v1/users.labels#Label + :param label_name: The display name of the label. + :param label_visibility: Whether, or not, you would like to show the label in the web UI. + :param message_visibility: Whether, or not, you would like to show messages with this label in the web UI. + :param label_text_color: The colour of the text of the label. + :param label_background_color: The colour of the background/tag of the label. + :param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be + suppressed. + :return: True if the label was created, else False. + """ + + # Start by assuming failure: + success = False + + try: + + # Standard token-refresh check: + await self.__ensure_token() + + # Make the API call: + api_response = await self.__http_client.post( + url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/labels", + headers = { + "Authorization": f"Bearer {self.__credentials.token}" + }, + json = { + "name": label_name, + "messageListVisibility": "show" if message_visibility else "hide", + "labelListVisibility": "labelShow" if label_visibility else "labelHide", + "color": { + "textColor": label_text_color.lower(), + "backgroundColor": label_background_color.lower() + } + } + ) + + # If the API call failed: + if api_response.status_code not in [200]: return success + + # Else we note down the success: + if api_response.json().get("id") is not None: success = True + + # In case something goes wrong along the way: + except Exception as exception: + if raise_exception: raise + self._printer(exception) + success = False + + # Done here: + return success + + async def update_label( + self, + label_id: str, + label_name: str = None, + label_visibility: Literal["labelShow", "labelShowIfUnread", "labelHide"] = None, + message_visibility: Literal["show", "hide"] = None, + label_text_color: str = None, + label_background_color: str = None, + raise_exception: bool = False + ) -> bool: + + """ + Updates one label for the user. + DOCUMENTATION: + 1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/create + 2. https://developers.google.com/gmail/api/reference/rest/v1/users.labels#Label + NOTE: Both or none of the colours must be updated. For this reason, a simple default will be chosen for the + other if only one is provided. + :param label_id: The id that Google assigned to the label. + :param label_name: The display name of the label. + :param label_visibility: Whether, or not, you would like to show the label in the web UI. + :param message_visibility: Whether, or not, you would like to show messages with this label in the web UI. + :param label_text_color: The colour of the text of the label. + :param label_background_color: The colour of the background/tag of the label. + :param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be + suppressed. + :return: True if the label was created, else False. + """ + + # Start by assuming failure: + success = False + + try: + + # Standard token-refresh check: + await self.__ensure_token() + + # Format the JSON body: + json_body = {} + if label_name: json_body["name"] = label_name + if label_visibility: json_body["labelListVisibility"] = label_visibility + if message_visibility: json_body["messageListVisibility"] = message_visibility + if label_text_color or label_background_color: + json_body["color"] = { + "textColor": (label_text_color or "#434343").lower(), + "backgroundColor": (label_background_color or "#000000").lower() + } + + # If no value was given to update: + if not json_body: return success + + # Make the API call: + api_response = await self.__http_client.put( + url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/labels/{label_id}", + headers = {"Authorization": f"Bearer {self.__credentials.token}"}, + json = json_body + ) + + # If the API call failed: + if api_response.status_code not in [200]: return success + + # Else we note down the success: + if api_response.json().get("id") is not None: success = True + + # In case something goes wrong along the way: + except Exception as exception: + if raise_exception: raise + self._printer(exception) + success = False + + # Done here: + return success + + async def delete_label( + self, + label_id: str, + raise_exception: bool = False + ) -> bool: + + """ + To delete one label of this user. the label will be identified by its id. + DOCUMENTATION: + 1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/get + 2. https://developers.google.com/gmail/api/reference/rest/v1/users.labels#Label + :param label_id: The id that Google assigned to the label. + :param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be + suppressed. + :return: The list of labels. + """ + + # Start by assuming failure: + success = False + + try: + + # Standard token-refresh check: + await self.__ensure_token() + + # Make the API call: + api_response = await self.__http_client.delete( + url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/labels/{label_id}", + headers = {"Authorization": f"Bearer {self.__credentials.token}"} + ) + + # If the API call failed: + if api_response.status_code not in [200, 204]: return success + + # Else we extract the response: + else: success = True + + # In case something goes wrong along the way: + except Exception as exception: + if raise_exception: raise + self._printer(exception) + success = False + + # Done here: + return success + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + import dateparser + + # Create an HTTP client: + test_client = httpx.AsyncClient( + limits = httpx.Limits( + max_connections = 100, # ............ Maximum number of connections allowed in the pool. + max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive. + ), + timeout = httpx.Timeout( + connect = 2.5, # ... Shorter connection timeout. + read = 2.5, # ...... Like what EasyEcom gives. + write = 10.0, # .... Time to wait for sending data. + pool = 120.0 # ..... Time to wait for a free connection from the pool. + ) + ) + + secrets_file = r"../../creds/google_tcaoff_test_oauth_20241125.json" + secrets_dict = json.from_file(secrets_file) + + my_oauth = GoogleOAuth( + config = secrets_dict, + redirect_url = r"https://api.thecaoffice.com/converse/mail/callback/gmail", + debug = True, + debug_prefix = "OAuth (Goog) | " + ) + + tokens = { + "access_token": "ya29.a0AeDClZAYoo85BXRId_n-hwo_amKshzi46c33GaJcsZZvGB7A7OGU2RFYcWBM_BleNBfAFUSJP2NHAvmd7Nsp_U5Kg68hXSy0iO99PNTm3pvKrJSzbkA-rXsVLsCnBIfPUMyNt2nOOVJmGwm17DNN0jAELkm1fPNTju7SZzmuaCgYKAZwSARMSFQHGX2Mis8TZui2rZT1gKySVds-N0w0175", + "refresh_token": "1//0gnqzjMf9YT19CgYIARAAGBASNgF-L9Ir3rcY37nGrV45XyOUBRllEH7Txui7T1JbwevlmDoNw7PuMu149cCWQSwsScuKaZusUQ", + "expires_in": 3539, + "expires_at": dateparser.parse("2024-11-25 10:40:40.833699+00:00"), + "scopes": [ + "https://www.googleapis.com/auth/gmail.labels", + "https://www.googleapis.com/auth/gmail.modify" + ] + } + + async def main(): + + my_gmail = AsyncGMailClient( + http_client = test_client, + credentials = await my_oauth.credentials_from_tokens(tokens = tokens) + ) + await my_gmail.initialize() + print(await my_gmail.user_email) + print(await my_gmail.user_profile) + + # print("Listing labels.") + # print(json.to_string(await my_gmail.list_labels())) + + # print("Getting one label.") + # print(json.to_string(await my_gmail.get_label(label_id = "Label_3"))) + + # print("Updating one label.") + # print(json.to_string(await my_gmail.update_label( + # label_id = "Label_3", + # label_name = "Updated Label 123", + # label_background_color = "#cc3a21" + # ))) + # + # print("Getting one label.") + # print(json.to_string(await my_gmail.get_label(label_id = "Label_3"))) + + # print("Creating label:") + # print(json.to_string(await my_gmail.create_label( + # label_name = "Bye - Bye !", + # label_text_color = "#fbc8d9", + # label_background_color = "#7a2e0b" + # ))) + + print("Deleting one label.") + print(json.to_string(await my_gmail.delete_label(label_id = "Label_6"))) + + + asyncio.run(main()) diff --git a/utils_v2/oauth/services/goog.py b/utils_v2/oauth/services/goog.py index 141132f..2a66191 100644 --- a/utils_v2/oauth/services/goog.py +++ b/utils_v2/oauth/services/goog.py @@ -110,6 +110,7 @@ SCOPES_GMAIL_FULL = [r"https://mail.google.com/"] class GoogleOAuth(OAuthBase): # Class variables: + service_name = "google" __flow = None # ┏┓┓ ┳┳┓ ┓ ┓ @@ -291,6 +292,25 @@ class GoogleOAuth(OAuthBase): # ┗┛┗ ┛ ┗┛┗┗┗ ┗┛┣┛┗ ┗┗┛┗┗ # ┛ + @staticmethod + async def credentials_have_expired(credentials: Credentials) -> bool: + + """ + Checks is a 'Credentials' object has expired or not. The native mechanism has some complication with timezone + considerations. This method was created to overcome that bug. + :param credentials: The 'Credentials' object of the user. + :return: True if the credentials have expired, False if they're yet valid. + """ + + # Take the timezone into account: + expires_at = date_time.as_if_timezone( + credentials.expiry, + timezone = date_time.TIMEZONE_UTC + ) + + # Return the assessment based on the evaluation with the timezone considered: + return True if date_time.get_current_utc_date_time() >= expires_at else False + async def credentials_from_tokens( self, tokens: dict, @@ -355,26 +375,27 @@ if __name__ == "__main__": async def main(): await my_goog.initialize(scopes = SCOPES_GMAIL_MAIL_MANAGEMENT) - await asyncio.sleep(1.0) - print("AUTH URL 0:", await my_goog.get_authorization_url( - user_id = "BHOPLI", - # email = "pskhushal@gmail.com" - )) - redirect_url = input("Paste the redirect URL here: ") - tokens = await my_goog.get_tokens(redirect_url = redirect_url) - print("TOKENS:", json.to_string(tokens, default = str)) + # await asyncio.sleep(1.0) + # print("Service Name:", my_goog.service_name) + # print("AUTH URL 0:", await my_goog.get_authorization_url( + # user_id = "BHOPLI", + # # email = "pskhushal@gmail.com" + # )) + # redirect_url = input("Paste the redirect URL here: ") + # tokens = await my_goog.get_tokens(redirect_url = redirect_url) + # print("TOKENS:", json.to_string(tokens, default = str)) - # # To test refreshing: - # tokens = { - # "access_token": "ya29.a0AeDClZAYoo85BXRId_n-hwo_amKshzi46c33GaJcsZZvGB7A7OGU2RFYcWBM_BleNBfAFUSJP2NHAvmd7Nsp_U5Kg68hXSy0iO99PNTm3pvKrJSzbkA-rXsVLsCnBIfPUMyNt2nOOVJmGwm17DNN0jAELkm1fPNTju7SZzmuaCgYKAZwSARMSFQHGX2Mis8TZui2rZT1gKySVds-N0w0175", - # "refresh_token": "1//0gnqzjMf9YT19CgYIARAAGBASNgF-L9Ir3rcY37nGrV45XyOUBRllEH7Txui7T1JbwevlmDoNw7PuMu149cCWQSwsScuKaZusUQ", - # "expires_in": 3539, - # "expires_at": dateparser.parse("2024-11-25 10:40:40.833699+00:00"), - # "scopes": [ - # "https://www.googleapis.com/auth/gmail.labels", - # "https://www.googleapis.com/auth/gmail.modify" - # ] - # } + # To test refreshing: + tokens = { + "access_token": "ya29.a0AeDClZAYoo85BXRId_n-hwo_amKshzi46c33GaJcsZZvGB7A7OGU2RFYcWBM_BleNBfAFUSJP2NHAvmd7Nsp_U5Kg68hXSy0iO99PNTm3pvKrJSzbkA-rXsVLsCnBIfPUMyNt2nOOVJmGwm17DNN0jAELkm1fPNTju7SZzmuaCgYKAZwSARMSFQHGX2Mis8TZui2rZT1gKySVds-N0w0175", + "refresh_token": "1//0gnqzjMf9YT19CgYIARAAGBASNgF-L9Ir3rcY37nGrV45XyOUBRllEH7Txui7T1JbwevlmDoNw7PuMu149cCWQSwsScuKaZusUQ", + "expires_in": 3539, + "expires_at": dateparser.parse("2024-11-25 10:40:40.833699+00:00"), + "scopes": [ + "https://www.googleapis.com/auth/gmail.labels", + "https://www.googleapis.com/auth/gmail.modify" + ] + } new_tok = await my_goog.refresh_tokens(old_tokens = tokens, force_refresh = False) print("REFRESHED TOKENS:", json.to_string(new_tok, default = str))