""" AUTHOR: Omkar Khandare DATE: Friday, 27th Jun., 2025. OBJECTIVE: To handle ecommerce authentication. 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.string import json 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.core.software import CoreSoftwareController # To make very controlled API calls: from utils_v2.rest.controllers.async_base import AsyncREST from utils_v2.rest.models.api_call import ApiResponse # Models: # Models: from models.core.user import CoreUserInfoModel from models.core.auth_token import CoreAuthTokenModel from models.core.user import CoreUserInfoModel from models.api.message.mail.oauth import ( OAuthMailAuthorizationRequestHeaders, OAuthMailAuthorizationRequestData ) from models.message.mail.oauth import OAuthMailGetAuthorizationURLResponse, OAuthMailHandleCallbackResponse from models.core.message import CoreMessageModel # To make HTTP requests: import httpx # to work with MongoDB: from bson.objectid import ObjectId # Mail Client(s): from utils_v2.goog.controllers.places.places_client import AsyncPlacesClient # To make abstract classes: from abc import ABC, abstractmethod # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class GooglePlacesController(CoreSoftwareController, ABC): # ┏┓┓ ┓┏ # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ SERVICE_TYPE = "software" # ┏┓ # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ def __init__( self, cache: AsyncRedisCache = None, http_client: httpx.AsyncClient = None, alert_url: str = None, base_filter: dict = None, debug: bool = True, debug_prefix: str = "Google Places (C) | ", debug_only_errors: bool = True ): """ This is the foundational controller for generate an authentication key for ecommerce integration, To Validate user from ecommerce webhooks :param cache: The object to use for caching results from database calls. :param http_client: The HTTP client to use to make REST-ful API calls. :param base_filter: The basic filter that will be applied to all fetching/updating queries. WARNING: THE BASE FILTER WILL ALWAYS BE APPLIED AUTOMATICALLY. SET THIS UP WISELY. :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. """ # Prepare the combined base filter: shopify_filter = {} for k, v in (base_filter or {}).items(): shopify_filter[k] = v shopify_filter["serviceType"] = self.SERVICE_TYPE # Invoke the parent's constructor: CoreSoftwareController.__init__( self, cache = cache, alert_url = alert_url, http_client = http_client, base_filter = shopify_filter, debug = debug, debug_prefix = debug_prefix, debug_only_errors = debug_only_errors ) # Init a variable in a parent: self._service_type = self.SERVICE_TYPE # For controlled REST-ful calls: self._rest = AsyncREST( http_client = http_client, debug = debug, debug_prefix = debug_prefix, debug_only_errors = debug_only_errors ) # ┏┓┏┓ ┓ ┏┓ ┏┓ # ┃┃┣┫┓┏╋┣┓┏┛ ┃┫ # ┗┛┛┗┗┻┗┛┗┗━•┗┛ @abstractmethod async def get_authorization_url( self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, mail_client: AsyncPlacesClient, 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. """ pass @abstractmethod async def handle_authorization_callback( self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, mail_client: AsyncPlacesClient, request_url: str, inbound_data: dict, session_token: str = None ) -> OAuthMailHandleCallbackResponse: """ To handle the authorization callback for the mail client. The user may grant or deny authorization. :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 request_url: The full callback URL invoked by the third-party client. :param inbound_data: The data that came in with the request (API call). :param session_token: The session token of the user. It is expected that this will be null in all cases. :return: A structured response of the process of handling the mail callback. """ pass @abstractmethod async def refresh_authorization( self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, mail_client: AsyncPlacesClient, http_client: httpx.AsyncClient, auth_token: CoreAuthTokenModel, force_refresh: bool = False, session_token: str = None ) -> CoreAuthTokenModel: """ To refresh the third-party client's access/authorization token(s) before use. :param sql_conn: The database connection to use when storing the refreshed tokens. :param mongo_data_conn: The database connection to use when storing the refreshed tokens. :param mail_client: The connection of the third-party mail client. :param http_client: The HTTP client to use to make the token refresh request. :param auth_token: The auth-token model of the existing integration. This may get updated if a refresh is needed (or forced). :param force_refresh: Whether, or not, you would like to force a refresh even if the token hasn't expired yet. :param session_token: The session token of the user. This will be null if this method is invoked by a cron script in the background. Needed only to identify the user in case of a failure to send a timely alert. :return: The same auth-token model instance, but maybe with updated tokens. """ pass # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass