""" AUTHOR: Khushal P Soonderji DATE: Thursday, 16th Jan., 2025. OBJECTIVE: To handle all mail-related behaviour for Gmail from one place. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # My async utils: from utils_v2.date_time import date_time from utils_v2.database.async_mysql_v2 import AsyncMySQL from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache # Controllers: from controllers_v2.message.mail.base import MailController # Models: from models.core.user import CoreUserInfoModel from models.core.auth_token import CoreAuthTokenModel from models.core.message import CoreMessageModel from models.api.message.mail.oauth import ( OAuthMailAuthorizationRequestHeaders, OAuthMailAuthorizationRequestData ) from models.message.mail.oauth import OAuthMailGetAuthorizationURLResponse, OAuthMailHandleCallbackResponse # Mail Client(s): from utils_v2.goog.controllers.gmail.gmail_client import AsyncGMailClient, SCOPES_GMAIL_MAIL_MANAGEMENT # To work with datatypes: from typing import List, Any # To make HTTP requests: import httpx # For asynchronous activities: import asyncio # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class GmailController(MailController): # ┏┓ # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ def __init__( self, cache: AsyncRedisCache = None, http_client: httpx.AsyncClient = None, alert_url: str = None, debug: bool = True, debug_prefix: str = "Gmail (C) | ", debug_only_errors: bool = True ): """ This is the controller specifically built for Gmail's services. It is built on top of the base mail controller. :param cache: The object to use for caching results from database calls. :param http_client: The HTTP client :param debug: Whether, or not, you would like to print debugging messages: :param debug_prefix: The prefix to print with the debugging messages. :param debug_only_errors: Whether you would like to print only error messages or all messages. :return: None. """ # Invoke the parent's constructor: super().__init__( cache = cache, alert_url = alert_url, http_client = http_client, base_filter = {"client": "gmail"}, debug = debug, debug_prefix = debug_prefix, debug_only_errors = debug_only_errors ) # ┏┓┏┓ ┓ ┏┓ ┏┓ # ┃┃┣┫┓┏╋┣┓┏┛ ┃┫ # ┗┛┛┗┗┻┗┛┗┗━•┗┛ async def get_authorization_url( self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, mail_client: AsyncGMailClient, user_info: CoreUserInfoModel, inbound_data: OAuthMailAuthorizationRequestData, session_token: str ) -> OAuthMailGetAuthorizationURLResponse: """ To accept an incoming request for mail integration and provide a URL that the user can use to authorize your service to access his mail inbox. :param sql_conn: The database connection to use to perform this task. :param mongo_data_conn: The database connection to use to perform this task. :param mail_client: The instance of the third-party mail client that will be used to get the URL. :param user_info: The information about your user who is trying to use this system. :param inbound_data: The data that came in with the request (API call). :param session_token: The session token of the user. :return: A structure response with details about the URL generation process. """ # Start by assuming failure: response = OAuthMailGetAuthorizationURLResponse() # First, we create/update a record for this integration request: token_key = await self.generate_token_key( sql_conn = sql_conn, mongo_data_conn = mongo_data_conn, auth_token = CoreAuthTokenModel( serviceType = "email", client = inbound_data.mailClient, authType = "oauth", user = user_info, clientUserId = {"email": inbound_data.mailId}, status = "pending", syncFreq = inbound_data.syncFreq, ), token_notes = { "email": inbound_data.mailId, "client": inbound_data.mailClient }, display_name = inbound_data.mailId, display_picture = None, session_token = session_token ) # If generating the token key fails: if token_key is None: response.message = "Failed to generate token key." return response # Now we create the URL: response.url = await mail_client.get_authorization_url( scopes = SCOPES_GMAIL_MAIL_MANAGEMENT, state = str(token_key), access_type = "offline", approval_prompt = "force", include_granted_scopes = "true", user_email = inbound_data.mailId ) response.success = True response.message = "Please use the URL to integrate your Gmail account." # Done here: return response async def handle_authorization_callback( self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, mail_client: AsyncGMailClient, request_url: str, inbound_data: dict, session_token: str = None ) -> OAuthMailHandleCallbackResponse: # Start by assuming failure: response = OAuthMailHandleCallbackResponse() # In case the user denied access: if inbound_data.get("error") == "access_denied": response.action = "denied" response.message = "The user denied authorization." return response # Otherwise we know that the user authorized access: else: response.action = "authorized" response.message = "The user has given authorization." pass # We fetch the auth-token associated with this authorization loop: auth_token = await self.get_token_from_key( mongo_data_conn = mongo_data_conn, token_key = inbound_data["state"] ) if not auth_token: response.message = "Failed to load the auth-token for this flow." return response # Generate the tokens from the callback. Google sends all the needed params in the callback as the URL's query # params. We can simply use the exact URL that was hit to generate the tokens. In Quart (and Flask) this can be # achieved by 'request.url' like this: google_tokens = await mail_client.get_authorization_tokens( redirect_url = request_url, scopes = None ) # If no tokens were generated: if not google_tokens: response.message = "Failed to get access token(s) from Gmail." return response # Try getting the user's profile from Gmail: user_profile = await mail_client.get_user_profile(tokens = google_tokens) if user_profile.success: google_tokens.email = user_profile.data["emailAddress"] google_tokens.displayName = user_profile.data["displayName"] google_tokens.displayPictureUrl = user_profile.data["displayPictureUrl"] else: response.message = "Failed to get the user's profile from Gmail." return response # We confirm if the expected email account and the one that gave authorization are the same: if auth_token.clientUserId["email"] != str(google_tokens.email): response.message = ( f"We were expecting authorization from '{auth_token.clientUserId['email']}', " f"but got authorization from '{google_tokens.email}' instead." ) return response # Now we try to create the standard set of Labels: labels = [ { "name": "TCAOFF", "textColor": "#434343", "backgroundColor": "#e7e7e7" }, { "name": "CA-Doc", "textColor": "#434343", "backgroundColor": "#e7e7e7" }, { "name": "CA-AI", "textColor": "#434343", "backgroundColor": "#e7e7e7" } ] tasks = [ mail_client.create_label( tokens = google_tokens, label_name = label["name"], label_visibility = "labelShow", message_visibility = "show", label_text_color = label["textColor"], label_background_color = label["backgroundColor"] ) for label in labels ] client_responses = await asyncio.gather(*tasks) # Add the labels to the tokens data: client_response = await mail_client.list_labels(tokens = google_tokens) google_tokens.labels = client_response.data if client_response.success else None # Now that we have passed the check, # we save the tokens to the database: auth_token.clientUserId = google_tokens.client_user_id auth_token.token = google_tokens.model_dump() auth_token.status = "active" tokens_saved = await self.set_token( sql_conn = sql_conn, mongo_data_conn = mongo_data_conn, token_key = inbound_data["state"], auth_token = auth_token, token_notes = { "email": google_tokens.email, "client": auth_token.client }, display_name = google_tokens.displayName, display_picture = google_tokens.displayPictureUrl, session_token = session_token ) # Note down the final result: if tokens_saved: response.success = True response.message = "Authorization flow completed successfully." else: response.message = "Failed to save the token(s)." # Done here: return response # ┳┳┓ •┓ ┏┓ • • # ┃┃┃┏┓┓┃ ┗┓┓┏┏┳┓┏┳┓┏┓┏┓┓┓┏┓╋┓┏┓┏┓ # ┛ ┗┗┻┗┗ ┗┛┗┻┛┗┗┛┗┗┗┻┛ ┗┗┗┻┗┗┗┛┛┗ pass # ┳┳┓ •┓ ┏┓ ╹• # ┃┃┃┏┓┓┃ ┗┓┓┏┏┓┏ ┓┏┓┏┓ # ┛ ┗┗┻┗┗ ┗┛┗┫┛┗┗ ┗┛┗┗┫ # ┛ ┛ # To synchronize the mails on the third-party client's server and your server. You are effectively making a copy of # the mail on your database. pass # ┳┳┓ •┓ ┓ • • # ┃┃┃┏┓┓┃ ┃ ┓┏╋┓┏┓┏┓ # ┛ ┗┗┻┗┗ ┗┛┗┛┗┗┛┗┗┫ # ┛ # Use these to show your users their mails once the mails are on your server. This would include activities like # listing mails, showing full mails, showing mail trails, etc. pass # ┳┳┓ •┓ ┏┓ ┓• # ┃┃┃┏┓┓┃ ┗┓┏┓┏┓┏┫┓┏┓┏┓ # ┛ ┗┗┻┗┗ ┗┛┗ ┛┗┗┻┗┛┗┗┫ # ┛ pass # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass