""" AUTHOR: Omkar Khandare DATE: Tuesday, 24th 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: from models.software.ecommerce.auth import ( ShopifyAuth, ShopifyAuthResponse ) from models.core.user import CoreUserInfoModel # To make HTTP requests: import httpx # to work with MongoDB: from bson.objectid import ObjectId # To make abstract classes: from abc import ABC, abstractmethod # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class EcommerceController(CoreSoftwareController, ABC): # ┏┓┓ ┓┏ # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ SERVICE_TYPE = "ecommerce" # ┏┓ # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ 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 = "Shopify (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 save_auth( self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, auth: ShopifyAuth, user: CoreUserInfoModel, session_token: str ) -> ShopifyAuthResponse: """ Checks if a particular set of incoming credentials give access to a valid server and then stores the credentials. :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 auth: The set of credentials as received from the UI/API. :return: A structured response to indicate what happened during authorization. """ pass # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass