""" 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 3. Messages: https://developers.google.com/gmail/api/reference/rest/v1/users.messages 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.string import regex from utils_v2.date_time import date_time from utils_v2.mail import mail_parser # My Google utils: from utils_v2.oauth.services.goog import GoogleOAuth from utils_v2.goog.base import AsyncGoogleBase from utils_v2.goog.models.data.auth_tokens import GoogleAuthTokens from utils_v2.goog.models.data.api_call import GoogleApiResponse # 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, List, Any # For debugging: from icecream import IceCreamDebugger # For computational help: import math # For base64 encoding: import base64 # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncGMailClient(AsyncGoogleBase): async def get_user_profile( self, tokens: GoogleAuthTokens, user_id: str = "me", ) -> GoogleApiResponse: """ To get the list of labels of this user. DOCUMENTATION: 1. https://developers.google.com/gmail/api/reference/rest/v1/users/getProfile :param tokens: The object that holds the access token to the service. :param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me". :return: A structured response where the list of labels will be in the 'data' variable. """ # Ensure that the tokens are valid: await tokens.arefresh( http_client = self._http_client, client_id = self._client_id, client_secret = self._client_secret, force_refresh = False ) # Make the API call: if not self._debug_only_errors: self._printer("Getting User Profile.") api_response = await self.get( url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/profile", headers = {"Authorization": f"Bearer {tokens.accessToken}"} ) # If the call was successful: if api_response.httpCode in [200]: api_response.data = await api_response.get_json() # Done here: return api_response # ┓ ┓ ┓ # ┃ ┏┓┣┓┏┓┃┏ # ┗┛┗┻┗┛┗ ┗┛ async def list_labels( self, tokens: GoogleAuthTokens, user_id: str = "me" ) -> GoogleApiResponse: """ 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 tokens: The object that holds the access token to the service. :param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me". :return: A structured response where the list of labels will be in the 'data' variable. """ # Ensure that the tokens are valid: await tokens.arefresh( http_client = self._http_client, client_id = self._client_id, client_secret = self._client_secret, force_refresh = False ) # Make the API call: if not self._debug_only_errors: self._printer("Listing All Labels.", user_id) api_response = await self.get( url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/labels", headers = {"Authorization": f"Bearer {tokens.accessToken}"} ) # If the call was successful: if api_response.httpCode in [200]: api_response.success = True api_json = await api_response.get_json() api_response.data = {label.pop("name"): label for label in api_json.get("labels", [])} # Done here: return api_response async def get_label( self, tokens: GoogleAuthTokens, label_id: str, user_id: str = "me" ) -> GoogleApiResponse: """ 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 tokens: The object that holds the access token to the service. :param label_id: The id that Google assigned to the label. :param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me". :return: A structured response where the list of labels will be in the 'data' variable. """ # Ensure that the tokens are valid: await tokens.arefresh( http_client = self._http_client, client_id = self._client_id, client_secret = self._client_secret, force_refresh = False ) # Make the API call: if not self._debug_only_errors: self._printer("Getting One Label.", user_id) api_response = await self.get( url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/labels/{label_id}", headers = {"Authorization": f"Bearer {tokens.accessToken}"} ) # If the call was successful: if api_response.httpCode in [200]: api_response.success = True api_response.data = await api_response.get_json() # Done here: return api_response async def create_label( self, tokens: GoogleAuthTokens, 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", user_id: str = "me" ) -> GoogleApiResponse: """ Create one label for the user. Doesn't apply it to any mail, just creates it. 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 tokens: The object that holds the access token to the service. :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 user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me". :return: A structured response where the list of labels will be in the 'data' variable. """ # Ensure that the tokens are valid: await tokens.arefresh( http_client = self._http_client, client_id = self._client_id, client_secret = self._client_secret, force_refresh = False ) # Make the API call: if not self._debug_only_errors: self._printer("Getting One Label.", user_id) api_response = await self.post( url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/labels", headers = {"Authorization": f"Bearer {tokens.accessToken}"}, 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 call was successful: if api_response.httpCode in [200]: api_response.success = True api_response.data = await api_response.get_json() # Done here: return api_response async def update_label( self, tokens: GoogleAuthTokens, 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, user_id: str = "me" ) -> GoogleApiResponse: """ Updates one label for the user. DOCUMENTATION: 1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/update 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 tokens: The object that holds the access token to the service. :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 user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me". :return: A structured response where the list of labels will be in the 'data' variable. """ # Ensure that the tokens are valid: await tokens.arefresh( http_client = self._http_client, client_id = self._client_id, client_secret = self._client_secret, force_refresh = False ) # 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() } # Make the API call: if not self._debug_only_errors: self._printer("Updating One Label.", user_id) api_response = await self.put( url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/labels/{label_id}", headers = {"Authorization": f"Bearer {tokens.accessToken}"}, json = json_body ) # If the call was successful: if api_response.httpCode in [200]: api_response.success = True api_response.data = await api_response.get_json() # Done here: return api_response async def delete_label( self, tokens: GoogleAuthTokens, label_id: str, user_id: str = "me" ) -> GoogleApiResponse: """ 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/delete :param tokens: The object that holds the access token to the service. :param label_id: The id that Google assigned to the label. :param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me". :return: A structured response where the list of labels will be in the 'data' variable. """ # Ensure that the tokens are valid: await tokens.arefresh( http_client = self._http_client, client_id = self._client_id, client_secret = self._client_secret, force_refresh = False ) # Make the API call: if not self._debug_only_errors: self._printer("Getting One Label.", user_id) api_response = await self.delete( url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/labels/{label_id}", headers = {"Authorization": f"Bearer {tokens.accessToken}"} ) # If the call was successful: if api_response.httpCode in [200, 204]: api_response.success = True # Done here: return api_response # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": import dateparser from utils_v2.goog.models.data.auth_tokens import GoogleAuthTokens # 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) tokens = { "accessToken": "ya29.a0AeDClZAYoo85BXRId_n-hwo_amKshzi46c33GaJcsZZvGB7A7OGU2RFYcWBM_BleNBfAFUSJP2NHAvmd7Nsp_U5Kg68hXSy0iO99PNTm3pvKrJSzbkA-rXsVLsCnBIfPUMyNt2nOOVJmGwm17DNN0jAELkm1fPNTju7SZzmuaCgYKAZwSARMSFQHGX2Mis8TZui2rZT1gKySVds-N0w0175", "refreshToken": "1//0gnqzjMf9YT19CgYIARAAGBASNgF-L9Ir3rcY37nGrV45XyOUBRllEH7Txui7T1JbwevlmDoNw7PuMu149cCWQSwsScuKaZusUQ", "expiresIn": 3539, "expiresAt": 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( service_name = "gmail", client_id = secrets_dict["web"]["client_id"], client_secret = secrets_dict["web"]["client_secret"], http_client = test_client, debug = True, debug_prefix = "GMail (M) | ", debug_only_errors = False ) test_tokens = GoogleAuthTokens(**tokens) response = await my_gmail.get_user_profile(tokens = test_tokens) print("RESPONSE:", response) asyncio.run(main())