From 485a8bd4868b000639111c9e22f73d1796912d74 Mon Sep 17 00:00:00 2001 From: khushal Date: Wed, 27 Nov 2024 16:54:08 +0530 Subject: [PATCH] (20241127) Google API OAuth2.0 support ready! --- utils_v2/goog/base.py | 95 ++++++- utils_v2/goog/gmail/gmail_client.py | 36 ++- utils_v2/goog/gmail/gmail_message.py | 20 +- utils_v2/mail/mail_parser.py | 44 ++- utils_v2/oauth/__init__.py | 0 utils_v2/oauth/base.py | 211 -------------- utils_v2/oauth/services/__init__.py | 0 utils_v2/oauth/services/goog.py | 405 --------------------------- 8 files changed, 177 insertions(+), 634 deletions(-) delete mode 100644 utils_v2/oauth/__init__.py delete mode 100644 utils_v2/oauth/base.py delete mode 100644 utils_v2/oauth/services/__init__.py delete mode 100644 utils_v2/oauth/services/goog.py diff --git a/utils_v2/goog/base.py b/utils_v2/goog/base.py index 1aef5d3..4e8b534 100644 --- a/utils_v2/goog/base.py +++ b/utils_v2/goog/base.py @@ -32,6 +32,9 @@ # To make sibling directories accessible for imports: import sys + +from utils_v2.goog.models.data.auth_tokens import GoogleAuthTokens + sys.path.append(".") sys.path.append("..") @@ -47,6 +50,7 @@ from utils_v2.goog.models.data.api_call import GoogleApiResponse from utils_v2.mail import mail_parser # Related to Google: +from google_auth_oauthlib.flow import InstalledAppFlow from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from googleapiclient.discovery import build @@ -116,9 +120,10 @@ class AsyncGoogleBase: def __init__( self, service_name: str, - client_id: str, - client_secret: str, + oauth_json: dict, http_client: httpx.AsyncClient, + scopes: List[str] = None, + redirect_url: str = None, debug = True, debug_prefix = "GMail | ", debug_only_errors = True @@ -128,8 +133,10 @@ class AsyncGoogleBase: To initialize any Google API from one base class. The client's id and secret are available in the file downloaded form https://console.cloud.google.com/apis/credentials (do not forget to select your app). :param service_name: A string to identify this service. - :param client_id: From the OAuth JSON downloaded from + :param oauth_json: The OAuth credentials downloaded from https://console.cloud.google.com/apis/credentials :param http_client: An asynchronous HTTP client to make API calls. + :param redirect_url: Where you would like to receive the confirmation of the user authorization. + :param scopes: The list of permissions needed for this particular authorization. :param debug: Whether, or not, you would like to show debugging messages on the terminal. :param debug_prefix: The prefix string to identify the debugging messages. :param debug_only_errors: Whether you would like to show all debugging messages or just error messages. @@ -143,9 +150,16 @@ class AsyncGoogleBase: # Accept the input configuration: self._service_name = service_name - self._client_id = client_id - self._client_secret = client_secret self._http_client = http_client + self._oauth_json = oauth_json + self._client_id = self._oauth_json["web"]["client_id"] + self._client_secret = self._oauth_json["web"]["client_secret"] + self._redirect_url = redirect_url + self._flow = InstalledAppFlow.from_client_config( + self._oauth_json, + scopes = scopes, + redirect_uri = self._redirect_url + ) def enable_debug(self): self._printer.enable() @@ -159,6 +173,76 @@ class AsyncGoogleBase: def debug_everything(self): self._debug_only_errors = False + # ┏┓┏┓ ┓ ┏┓ ┏┓ + # ┃┃┣┫┓┏╋┣┓ ┏┛ ┃┫ + # ┗┛┛┗┗┻┗┛┗ ┗━•┗┛ + + async def get_authorization_url( + self, + state: str = None, + access_type: Literal["online", "offline"] = "offline", + approval_prompt: Literal["auto", "force", "consent"] = "force", + include_granted_scopes: Literal["true", "false"] = "true", + user_email: str = None + ) -> str: + + """ + TO get the OAuth2.0 authorization URL for one user. + DOCUMENTATION: + 1. https://developers.google.com/identity/protocols/oauth2/web-server + :param state: A unique identifier for your user. If not supplied, a random string will be generated. + :param access_type: Set the value to offline if your application needs to refresh access tokens when the user is + not present at the browser. + :param approval_prompt: "force" ensures that the consent screen is always shown to the user, regardless of + whether the user has previously granted consent for the requested scopes. It forces the user to re-approve + the app’s access, which can be useful if the app is requesting new permissions or if the consent needs to be + explicitly confirmed. "consent" ensures the user’s consent is required if they haven't approved the app’s + requested permissions yet. "auto" allows Google to automatically determine whether the consent screen should + be shown. + :param include_granted_scopes: Enables applications to use incremental authorization to request access to + additional scopes in context. If you set this parameter's value to true and the authorization request is + granted, then the new access token will also cover any scopes to which the user previously granted the + application access. + :param user_email: + :return: + """ + + # Get an authorization URL: + auth_url, state = self._flow.authorization_url( + access_type = access_type, + approval_prompt = approval_prompt, + include_granted_scopes = include_granted_scopes, + login_hint = user_email, + state = state + ) + + # Done here: + return auth_url + + async def get_authorization_tokens( + self, + redirect_url: str + ) -> GoogleAuthTokens: + + """ + When the user accepts or declines an authorization request, Google sends you an alert on your redirect URL. Pass + the URL as it is to this method to generate the authorization tokens that you can store in the database and + reuse for this user's activities. + :param redirect_url: The exact URL that was hit (with the query params) that Google hit when the user did + something on your authorization URL. Fortunately, this URL is readily available in Quart and Flask by + calling 'request.url'. + :return: The authorization tokens. + """ + + credentials = self._flow.fetch_token(authorization_response = redirect_url) + ttl = credentials["expires_in"] - 60 + return GoogleAuthTokens( + accessToken = credentials.get("access_token"), + refreshToken = credentials.get("refresh_token"), + expiresAt = date_time.get_current_utc_date_time() + datetime.timedelta(seconds = ttl), + scopes = credentials.get("scope") + ) + # ┏┓ ┳┓ ┓• # ┣ ┏┓┏┓┏┓┏┓ ┃┃┏┓┏┏┓┏┫┓┏┓┏┓ # ┗┛┛ ┛ ┗┛┛ ┻┛┗ ┗┗┛┗┻┗┛┗┗┫ @@ -242,6 +326,7 @@ class AsyncGoogleBase: :param headers: The headers to pass. :param json: The params to send in the JSON body. :param data: The params to send in the form-data in the body. + :param content: The raw content to be sent in the body (typically as an octet-stream). :return: A structured response that includes the raw response, the exception (if any), and so on. """ diff --git a/utils_v2/goog/gmail/gmail_client.py b/utils_v2/goog/gmail/gmail_client.py index a837c11..c6c8248 100644 --- a/utils_v2/goog/gmail/gmail_client.py +++ b/utils_v2/goog/gmail/gmail_client.py @@ -34,6 +34,8 @@ # To make sibling directories accessible for imports: import sys +from http.client import responses + sys.path.append(".") sys.path.append("..") @@ -47,7 +49,6 @@ 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 @@ -88,7 +89,12 @@ import base64 # ***************************************************************************************************************** -# --- Nothing Yet +# Google Scopes: +SCOPES_GMAIL_MAIL_MANAGEMENT = [ + r"https://www.googleapis.com/auth/gmail.modify", + r"https://www.googleapis.com/auth/gmail.labels" +] +SCOPES_GMAIL_FULL = [r"https://mail.google.com/"] # ***************************************************************************************************************** @@ -618,7 +624,7 @@ class AsyncGMailClient(AsyncGoogleBase): if api_response.httpCode in [200]: api_response.success = True api_json = await api_response.get_json() - raw_message = base64.urlsafe_b64decode(api_json["raw"]) + raw_message = base64.urlsafe_b64decode(api_json["raw"]).decode() if return_raw: api_response.data = raw_message else: parsed_message = mail_parser.parse(raw_message) @@ -897,19 +903,33 @@ if __name__ == "__main__": # Create an instance of the client: my_gmail = AsyncGMailClient( service_name = "gmail", - client_id = secrets_dict["web"]["client_id"], - client_secret = secrets_dict["web"]["client_secret"], + oauth_json = secrets_dict, http_client = test_client, + redirect_url = r"https://api.thecaoffice.com/converse/mail/callback/gmail", + scopes = [ + r"https://www.googleapis.com/auth/gmail.modify", + r"https://www.googleapis.com/auth/gmail.labels" + ], debug = True, debug_prefix = "GMail (M) | ", debug_only_errors = False ) + # Request Auth: + # print("AUTH URL:", await my_gmail.get_authorization_url( + # state = "User123-Bhopli" + # )) + + # Get tokens from callback: + # print("TOKENS:", await my_gmail.get_authorization_tokens( + # redirect_url = r"https://api.thecaoffice.com/converse/mail/callback/gmail?state=User123-Bhopli..." + # )) + # Create a sample mail: my_mail = GMailMessage( from_email = "pskhushal@gmail.com", to_email = "orangebhopli@gmail.com", - subject = "Re: Bhopli is the best! (Thread Test)", + subject = "Bhopli is the best! (Parts Sequence Test)", cc_emails = None, bcc_emails = None ) @@ -936,7 +956,7 @@ if __name__ == "__main__": """ ) - my_mail.add_text("This is how you should pet her 👇") + my_mail.add_text("(3rd part - text) This is how you should pet her 👇") my_mail.add_inline_image(r"../../../data/images/cat_petting.png") my_mail.add_attachment(r"../../../data/pdf/sample_label.pdf") # print(my_mail.get_raw_message(as_base64 = False)) @@ -945,7 +965,7 @@ if __name__ == "__main__": response = await my_gmail.send_message( tokens = test_tokens, message = my_mail, - thread_id = "1936caffb67996d3" + thread_id = None ) print("SUCCESS:", response.success) print("SUMMARY:", response.to_markdown()) diff --git a/utils_v2/goog/gmail/gmail_message.py b/utils_v2/goog/gmail/gmail_message.py index bb29eb5..62556bd 100644 --- a/utils_v2/goog/gmail/gmail_message.py +++ b/utils_v2/goog/gmail/gmail_message.py @@ -180,12 +180,19 @@ class GMailMessage: self.message.attach(MIMEText(html_text, "html")) - def add_inline_image(self, image_file, content_id = None): + def add_inline_image( + self, + image_file: str | io.BytesIO, + file_name: str = None, + content_id: str = None + ): """ Add an inline image to the body of the mail. NOTE: This is NOT the same as sending an image as an attachment. :param image_file: The image data to attach to the mail body. + :param file_name: The name of the file. This is the same name by which it will be downloaded. You need not + specify this if the input file is specified as a path. Needed when you give the input file as a buffer. :param content_id: Inline images are inserted via HTML bocks. This field identifies the image resource. If not specified, I will generate a random string. You may write a custom value here if you know what you are doing. For most use cases, please ignore this field. @@ -195,6 +202,7 @@ class GMailMessage: # Read the image as bytes: image_bytes = None if type(image_file) is str: + file_name = file_name or os.path.split(image_file)[-1] with open(image_file, "rb") as opened_image_file: image_bytes = opened_image_file.read() if type(image_file) is io.BytesIO: @@ -217,7 +225,13 @@ class GMailMessage: # Then add the image: image_part = MIMEImage(image_bytes) - image_part.add_header("Content-ID", f"<{content_id}>") + image_part.add_header( + "Content-ID", f"<{content_id}>" + ) + image_part.add_header( + "Content-Disposition", + f"inline; filename=\"{file_name}\"", + ) self.message.attach(image_part) def add_attachment( @@ -253,7 +267,7 @@ class GMailMessage: encoders.encode_base64(part) part.add_header( "Content-Disposition", - f"attachment; filename= {file_name}", + f"attachment; filename=\"{file_name}\"", ) self.message.attach(part) diff --git a/utils_v2/mail/mail_parser.py b/utils_v2/mail/mail_parser.py index 8bcb85e..7229112 100644 --- a/utils_v2/mail/mail_parser.py +++ b/utils_v2/mail/mail_parser.py @@ -24,8 +24,7 @@ N/A """ - - +import base64 # ***************************************************************************************************************** # ***** **** # *** IMPORT *** @@ -49,6 +48,9 @@ from utils_v2.date_time import date_time # To work with mails: import mailparser +# To parse the HTML content in the mail: +from bs4 import BeautifulSoup + # To work with datatypes: from typing import Any, Dict @@ -111,6 +113,42 @@ def parse(raw_mail: str | bytes) -> Dict[str, Any]: } for attachment in parsed_mail.attachments ] + # Figure out which entity (text and HTML) came in which sequence. + # The library doesn't give us any sequence info so we do some custom string processing here to figure out the order + # in which to render the contents of the page. + parts = [ + { + "no": None, + "offset": max( + parsed_mail.message_as_string.find(t), + parsed_mail.message_as_string.find(base64.b64encode(t.encode()).decode()) + ), + "type": "text/plain", + "data": t + } for t in parsed_mail.text_plain + ] + parts = parts + [ + { + "partNo": None, + "offset": max( + parsed_mail.message_as_string.find(h), + parsed_mail.message_as_string.find(base64.b64encode(h.encode()).decode()) + ), + "type": "text/html", + "data": h + } for h in parsed_mail.text_html + ] + parts = sorted(parts, key = lambda x: x["offset"]) + for i, p in enumerate(parts): p["no"] = i + + # Get the unformatted text from everything in the mail: + unformatted_text = [] + for p in parts: + if p["type"] == "text/html": + html_parser = BeautifulSoup(p["data"], "html.parser") + unformatted_text.append(html_parser.get_text()) + else: unformatted_text.append(p["data"]) + # Put everything together: return { "ts": date_time.to_timezone(parsed_mail.date, timezone = date_time.TIMEZONE_UTC), @@ -121,6 +159,8 @@ def parse(raw_mail: str | bytes) -> Dict[str, Any]: "bcc": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers.get("Bcc", [])], "text": parsed_mail.text_plain, "html": parsed_mail.text_html, + "parts": parts, + "unformattedText": "\n".join(unformatted_text), "attachments": message_attachments, } diff --git a/utils_v2/oauth/__init__.py b/utils_v2/oauth/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/utils_v2/oauth/base.py b/utils_v2/oauth/base.py deleted file mode 100644 index 0f45c7d..0000000 --- a/utils_v2/oauth/base.py +++ /dev/null @@ -1,211 +0,0 @@ -""" - - AUTHOR: - - Khushal P Soonderji - - DATE: - - Friday, 22nd Nov., 2024 - - OBJECTIVE: - - To provide a structured way to handle OAuth2.0 behaviour for various services. - - REFERENCES: - - N/A - - DOWNLOADS: - - N/A - -""" - - -# ***************************************************************************************************************** -# ***** **** -# *** IMPORT *** -# ***** **** -# ***************************************************************************************************************** - - -# To make sibling directories accessible for imports: -import sys -sys.path.append(".") -sys.path.append("..") - -# System-level activities: -import io - -# For defining the class's structure: -from abc import ABC, abstractmethod - -# For working with datatypes: -from typing import List - -# For debugging: -from icecream import IceCreamDebugger - - -# ***************************************************************************************************************** -# ***** **** -# *** MACROS / ONE-TIME INIT *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** VARIABLES *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** FUNCTIONS *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** CLASSES *** -# ***** **** -# ***************************************************************************************************************** - - -class OAuthBase(ABC): - - def __init__( - self, - config: dict, - redirect_url: str, - debug = True, - debug_prefix = "OAuth | ", - 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 config: - self._config = config - self._redirect_url = redirect_url - - 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 - - # ┏┓┓ ┳┳┓ ┓ ┓ - # ┣┫┣┓┏╋┏┓┏┓┏╋ ┃┃┃┏┓╋┣┓┏┓┏┫┏ - # ┛┗┗┛┛┗┛ ┗┻┗┗ ┛ ┗┗ ┗┛┗┗┛┗┻┛ - - @abstractmethod - async def initialize( - self, - scopes: List, - raise_exception = False - ) -> bool: - - """ - To initialize the service-specific OAuth2.0 class. For example, in Google's case, we need to initialize an app - flow that was created for an app through its Cloud Console panel. - :param scopes: The list of permissions being requested. The word 'scopes' has been borrowed from Google's OAuth - documentation (which was implemented first). - :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any - exception that occurs will be suppressed. - :return: True if the initialization succeeded, False if it failed. - """ - - pass - - @abstractmethod - async def get_authorization_url( - self, - raise_exception = False, - **kwargs, - ) -> str | None: - - """ - To create an authorization URL which will be then sent to the front-end for the user to click and grant/decline - various permissions. - :param kwargs: The identifiers of the user who wants to use your service and any other service-specific options. - :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any - exception that occurs will be suppressed. - :return: The authorization URL if successful, or None if failed. - """ - - pass - - @abstractmethod - async def get_tokens( - self, - raise_exception = False, - **kwargs - ) -> dict | None: - - """ - To get the tokens of a user. Plural 'tokens' because OAuth typically has one access token that expires - every-so-often, and one refresh token that grants you a new access token. - :param kwargs: The identifiers of the user who wants to use your service and any other service-specific options. - :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any - exception that occurs will be suppressed. - :return: The tokens for the service if successful, or None if failed. - """ - - pass - - @abstractmethod - async def refresh_tokens( - self, - old_tokens: dict, - force_refresh = False, - raise_exception = False - ) -> dict | None: - - """ - To refresh the tokens of a user. Plural 'tokens' because OAuth typically has one access token that expires - every-so-often, and one refresh token that grants you a new access token. - :param old_tokens: The current set of tokens. - :param force_refresh: To force a refresh request even if the tokens haven't yet expired. - :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any - exception that occurs will be suppressed. - :return: The same tokens if they haven't expired, refreshed tokens if the tokens have expired and were - successfully refreshed, None if the tokens have expired but could not be refreshed. - """ - - pass - - -# ***************************************************************************************************************** -# ***** **** -# *** MAIN PROGRAM *** -# ***** **** -# ***************************************************************************************************************** - - -if __name__ == "__main__": - - pass diff --git a/utils_v2/oauth/services/__init__.py b/utils_v2/oauth/services/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/utils_v2/oauth/services/goog.py b/utils_v2/oauth/services/goog.py deleted file mode 100644 index 3376e1c..0000000 --- a/utils_v2/oauth/services/goog.py +++ /dev/null @@ -1,405 +0,0 @@ -""" - - AUTHOR: - - Khushal P Soonderji - - DATE: - - Friday, 22nd Nov., 2024 - - OBJECTIVE: - - To handle OAuth2.0 activities for Google's services. - - REFERENCES: - - 1. https://developers.google.com/calendar/api/quickstart/python - 2. https://developers.google.com/identity/protocols/oauth2/web-server#python - 3. https://www.youtube.com/watch?v=vQQEaSnQ_bs&t=940s&pp=ygUVb2F1dGgyIHB5dGhvbiB5b3V0dWJl - - 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 - -# The base model: -from utils_v2.oauth.base import OAuthBase - -# Related to Google: -from google.auth.transport.requests import Request -from google.oauth2.credentials import Credentials -from google_auth_oauthlib.flow import InstalledAppFlow - -# For asynchronous activities: -import asyncio - -# To make deep-copies: -import copy - -# To work with date and time: -import datetime - -# For working with datatypes: -from typing import List - - -# ***************************************************************************************************************** -# ***** **** -# *** MACROS / ONE-TIME INIT *** -# ***** **** -# ***************************************************************************************************************** - - -# Google Scopes: -SCOPES_GMAIL_MAIL_MANAGEMENT = [ - r"https://www.googleapis.com/auth/gmail.modify", - r"https://www.googleapis.com/auth/gmail.labels" -] -SCOPES_GMAIL_FULL = [r"https://mail.google.com/"] - - -# ***************************************************************************************************************** -# ***** **** -# *** VARIABLES *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** FUNCTIONS *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** CLASSES *** -# ***** **** -# ***************************************************************************************************************** - - -class GoogleOAuth(OAuthBase): - - # Class variables: - service_name = "google" - __flow = None - - # ┏┓┓ ┳┳┓ ┓ ┓ - # ┣┫┣┓┏╋┏┓┏┓┏╋ ┃┃┃┏┓╋┣┓┏┓┏┫┏ - # ┛┗┗┛┛┗┛ ┗┻┗┗ ┛ ┗┗ ┗┛┗┗┛┗┻┛ - - async def initialize( - self, - scopes: List, - raise_exception = False - ) -> bool: - - """ - Initialize the Google OAuth mechanism by creating an app-flow. This defines the app that you are trying to - deploy. You must create this app in Google's Cloud Platform's console. - :param scopes: The scopes (permissions) needed by this app. - :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any - exception that occurs will be suppressed. - :return: True if the initialization succeeded, False if it failed. - """ - - # Start by assuming success: - success = True - - try: - - # Initialize your Google App: - self._printer("Initializing flow.") - self.__flow = InstalledAppFlow.from_client_config( - self._config, - scopes = scopes, - redirect_uri = self._redirect_url - ) - - # If something goes wrong: - except Exception as exception: - self._printer(exception) - if raise_exception: raise - success = False - - # Done here: - return success - - async def get_authorization_url( - self, - raise_exception = False, - **kwargs, - ) -> str | None: - - """ - To generate an authorization URL that can be sent to the front end. When the - :param raise_exception: If set to True, any exception that occurs will be propagated. If set to False, any - exception that occurs will be suppressed. - :param kwargs: Any no. of keyword args that you might want to give to this specific service. - :return: - """ - - # Start by assuming failure: - authorization_url = None - - try: - - # Request a URL that will be sent to the user to request - # permissions to access their account: - authorization_url, _ = self.__flow.authorization_url( - access_type = kwargs.get("access_type", "offline"), - approval_prompt = kwargs.get("approval_prompt", "force"), - include_granted_scopes = kwargs.get("include_granted_scopes", "true"), - login_hint = kwargs.get("email"), - state = kwargs.get("user_id") - ) - - # Done here: - return authorization_url - - # If something goes wrong: - except Exception as exception: - self._printer(exception) - if raise_exception: raise - authorization_url = None - - # Done here: - return authorization_url - - async def get_tokens( - self, - raise_exception = False, - **kwargs - ) -> dict | None: - - """ - To get the tokens of a user. Plural 'tokens' because OAuth typically has one access token that expires - every-so-often, and one refresh token that grants you a new access token. In Google's case, their servers hit - your callback URL with some query params. You must furnish this whole URL right here under the kwarg - 'redirect_url'. Fortunately, this URL is readily available in Quart and Flask by calling 'request.url'. - :param kwargs: The identifiers of the user who wants to use your service and any other service-specific options. - :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any - exception that occurs will be suppressed. - :return: The tokens for the service if successful, or None if failed. - """ - - # Start by assuming failure: - tokens = None - - try: - - # Fetch the tokens: - credentials = self.__flow.fetch_token(authorization_response = kwargs["redirect_url"]) - tokens = { - "access_token": credentials.get("access_token"), - "refresh_token": credentials.get("refresh_token"), - "expires_in": (ttl := credentials["expires_in"] - 60), - "expires_at": date_time.get_current_utc_date_time() + datetime.timedelta(seconds = ttl), - "scopes": credentials.get("scope"), - } - - # If something goes wrong: - except Exception as exception: - self._printer(exception) - if raise_exception: raise - tokens = None - - # Done here: - return tokens - - async def refresh_tokens( - self, - old_tokens: dict, - force_refresh = False, - raise_exception = False - ) -> dict | None: - - """ - To refresh the tokens of a user. Plural 'tokens' because OAuth typically has one access token that expires - every-so-often, and one refresh token that grants you a new access token. - :param old_tokens: The current set of tokens as obtained by 'get_tokens'. - :param force_refresh: To force a refresh request even if the tokens haven't yet expired. - :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any - exception that occurs will be suppressed. - :return: The same tokens if they haven't expired, refreshed tokens if the tokens have expired and were - successfully refreshed, None if the tokens have expired but could not be refreshed. - """ - - # Start by assuming failure: - tokens = None - - try: - - # If the tokens haven't expired, just return the existing tokens back: - tokens_expired = True if date_time.get_current_utc_date_time() >= old_tokens["expires_at"] else False - if not tokens_expired and not force_refresh: return old_tokens - - # Construct the credentials and request a refresh: - credentials = await self.credentials_from_tokens(old_tokens) - if credentials.refresh_token: - credentials.refresh(Request()) - tokens = { - "access_token": credentials.token, - "refresh_token": credentials.refresh_token, - "expires_at": (exp_at := date_time.as_if_timezone( - credentials.expiry, - timezone = date_time.TIMEZONE_UTC - )), - "expires_in": (exp_at - date_time.get_current_utc_date_time()).total_seconds(), - "scopes": credentials.scopes, - } - - # If something goes wrong: - except Exception as exception: - self._printer(exception) - if raise_exception: raise - tokens = None - - # Done here: - return tokens - - # ┏┓ • ┏┓ •┏• - # ┗┓┏┓┏┓┓┏┓┏┏┓ ┗┓┏┓┏┓┏┓╋┓┏ - # ┗┛┗ ┛ ┗┛┗┗┗ ┗┛┣┛┗ ┗┗┛┗┗ - # ┛ - - @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 - ) - print("EXP. AT:", expires_at) - - # 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, - raise_exception = False - ) -> Credentials | None: - - """ - This is a very service-specific thing. When using Google's APIs, you will need to use their native 'Credentials' - object. This method allows you to quickly convert the output of 'get_tokens' and 'refresh_tokens' to that object - for easy use. - :param tokens: The output of either 'get_tokens' or 'refresh_tokens'. - :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any - exception that occurs will be suppressed. - :return: Google's API's native 'Credentials' object if successful, else None. - """ - - # Start by assuming failure: - credentials = None - - try: - - # Make a deep-copy and add some fields from the config: - tokens_copy = copy.deepcopy(tokens) - first_key = list(self._config.keys())[0] - tokens_copy["client_id"] = self._config.get(first_key, {}).get("client_id") - tokens_copy["client_secret"] = self._config.get(first_key, {}).get("client_secret") - - # Create the credentials: - credentials = Credentials.from_authorized_user_info(info = tokens_copy) - - # If something goes wrong: - except Exception as exception: - self._printer(exception) - if raise_exception: raise - credentials = None - - # Done here: - return credentials - - -# ***************************************************************************************************************** -# ***** **** -# *** MAIN PROGRAM *** -# ***** **** -# ***************************************************************************************************************** - - -if __name__ == "__main__": - - import dateparser - - secrets_file = r"../../../creds/goog/app/google_tcaoff_test_oauth_20241125.json" - secrets_dict = json.from_file(secrets_file) - - my_goog = GoogleOAuth( - config = secrets_dict, - redirect_url = r"https://api.thecaoffice.com/converse/mail/callback/gmail", - debug = True, - debug_prefix = "OAuth (Goog) | " - ) - - async def main(): - - await my_goog.initialize(scopes = SCOPES_GMAIL_MAIL_MANAGEMENT) - # 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" - ] - } - - new_tok = await my_goog.refresh_tokens(old_tokens = tokens, force_refresh = False) - print("REFRESHED TOKENS:", json.to_string(new_tok, default = str)) - - - asyncio.run(main())