""" AUTHOR: Khushal P Soonderji DATE: Tuesday, 26th Nov., 2024 OBJECTIVE: To provide a base class for common behaviour of Google's APIs. REFERENCES: 1. Quickstart: https://developers.google.com/gmail/api/quickstart/python DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # 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("..") # 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.oauth.services.goog import GoogleOAuth 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 # 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 import inspect # For computational help: import math # For base64 encoding: import base64 # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncGoogleBase: def __init__( self, service_name: 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 ): """ 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 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. """ # 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._service_name = service_name 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() 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 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") ) # ┏┓ ┳┓ ┓• # ┣ ┏┓┏┓┏┓┏┓ ┃┃┏┓┏┏┓┏┫┓┏┓┏┓ # ┗┛┛ ┛ ┗┛┛ ┻┛┗ ┗┗┛┗┻┗┛┗┗┫ # ┛ @staticmethod async def __get_error_message(api_response: GoogleApiResponse) -> str: """ To extract various kinds of error messages from Google's responses. :param api_response: The formatted response from the API call. :return: The message string. """ try: return (await api_response.get_json())["error"]["message"] except: return api_response.response.reason_phrase # ┏┓┏┓┳ ┏┓ ┓┓• # ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓ # ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫ # ┛ async def get( self, url: str, headers: dict = None, params: dict = None ) -> GoogleApiResponse: """ To call an API using the GET method. :param url: The URL to call. :param headers: The headers to pass. :param params: The params to send in the query string itself. :return: A structured response that includes the raw response, the exception (if any), and so on. """ # Prepare the structure of the response: api_response = GoogleApiResponse( serviceName = self._service_name, action = inspect.stack()[1].function, url = url, method = "GET" ) try: # Make the API call: response = await self._http_client.get( url = url, headers = headers, params = params ) # Note down the results: api_response.response = response api_response.httpCode = response.status_code api_response.message = await self.__get_error_message(api_response) # If something goes wrong: except Exception as exception: api_response.exception = exception api_response.message = str(exception) self._printer(exception, api_response.url, api_response.method, headers, params) # Done here: return api_response async def post( self, url: str, headers: dict = None, json: dict = None, data: dict = None, content: str | bytes = None ) -> GoogleApiResponse: """ To call an API using the POST method. :param url: The URL to call. :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. """ # Prepare the structure of the response: api_response = GoogleApiResponse( serviceName = self._service_name, action = inspect.stack()[1].function, url = url, method = "POST" ) try: # Make the API call: response = await self._http_client.post( url = url, headers = headers, json = json, data = data, content = content ) # Note down the results: api_response.response = response api_response.httpCode = response.status_code api_response.message = await self.__get_error_message(api_response) # If something goes wrong: except Exception as exception: api_response.exception = exception api_response.message = str(exception) self._printer(exception, api_response.url, api_response.method, headers, json, data) # Done here: return api_response async def put( self, url: str, headers: dict = None, json: dict = None, data: dict = None ) -> GoogleApiResponse: """ To call an API using the PUT method. :param url: The URL to call. :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. :return: A structured response that includes the raw response, the exception (if any), and so on. """ # Prepare the structure of the response: api_response = GoogleApiResponse( serviceName = self._service_name, action = inspect.stack()[1].function, url = url, method = "PUT" ) try: # Make the API call: response = await self._http_client.put( url = url, headers = headers, json = json, data = data ) # Note down the results: api_response.response = response api_response.httpCode = response.status_code api_response.message = await self.__get_error_message(api_response) # If something goes wrong: except Exception as exception: api_response.exception = exception api_response.message = str(exception) self._printer(exception, api_response.url, api_response.method, headers, json, data) # Done here: return api_response async def delete( self, url: str, headers: dict = None ) -> GoogleApiResponse: """ To call an API using the DELETE method. :param url: The URL to call. :param headers: The headers to pass. :return: A structured response that includes the raw response, the exception (if any), and so on. """ # Prepare the structure of the response: api_response = GoogleApiResponse( serviceName = self._service_name, action = inspect.stack()[1].function, url = url, method = "DELETE" ) try: # Make the API call: response = await self._http_client.delete( url = url, headers = headers ) # Note down the results: api_response.response = response api_response.httpCode = response.status_code api_response.message = await self.__get_error_message(api_response) # If something goes wrong: except Exception as exception: api_response.exception = exception api_response.message = str(exception) self._printer(exception, api_response.url, api_response.method, headers) # Done here: return api_response # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass