""" AUTHOR: Khushal P Soonderji DATE: Friday, 30th Aug., 2024 OBJECTIVE: To be able to access SQL-based databases from python in a simple way. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # MySQL Database: import aiomysql # For data-crunching: import pandas as pd # For time-keeping: import time # OS-level operations: import os # My utils: from utils_v2.string import json # For async activities: import asyncio # For debugging: from icecream import IceCreamDebugger import traceback # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncMySQL: def __init__( self, pool_size, *args, **kwargs ): """ A class to work with SQL-based databases. Originally meant to only invoke stored procedures and retrieve them as JSON-like structures (list or dict). The format for the results was very specific to our use case for serving Bicree's requirement. This may not serve your requirement at all. :param pool_size: The number of connections to maintain n a pool. :param args: Any arguments to pass. Not used. :param kwargs: Pass the connection configuration from here. """ # Set up the variables: self.__args = args self.__kwargs = kwargs self.__min_pool_size = 10 self.__max_pool_size = max(pool_size, self.__min_pool_size) self.__pool = None # Set up the debugging tools: self.__printer = IceCreamDebugger(prefix = "MySQL | ", includeContext = True) def __del__(self): pass async def connect(self): """ Establish a connection and create a pool of connections to call from. :return: None. """ try: self.__kwargs["db"] = self.__kwargs.pop("database") self.__pool = await aiomysql.create_pool( minsize = self.__min_pool_size, maxsize = self.__max_pool_size, loop = asyncio.get_event_loop(), **self.__kwargs ) except Exception as exception: self.__printer(exception) self.__pool = None async def ensure_connection(self): """ Tries to ensure that a connection is present. Can be called before every function to make sure that our pool is established. :return: None. """ if self.__pool is None: await self.connect() async def call_procedure(self, procedure_name, procedure_args): """ To call stored procedures and retrieve all the responses. :param procedure_name: :param procedure_args: :return: """ # Make sure we have a connection: await self.ensure_connection() # Make a variable to hold all the result sets. # Needed for when the procedure responds with many "tables": all_result_sets = [] # Call the procedure and get the results: async with self.__pool.acquire() as connection: async with connection.cursor() as cursor: await cursor.callproc(procedure_name, procedure_args) # Iterate over all result sets, # and process them one-by-one: while True: this_result_set = [] result = await cursor.fetchall() if not result: break columns = [desc[0] for desc in cursor.description] for row in result: this_result_set.append(dict(zip(columns, row))) all_result_sets.append(this_result_set) await cursor.nextset() # Done here: return all_result_sets async def call_procedure_and_get_json( self, procedure_name, procedure_args, retry_count = 1, backoff_seconds = 0.5, backoff_multiplier = 1.1, return_exception = False ): # Note down the start time: start_ts = time.perf_counter() # Try to get the data from the database: results = [] exception = None for _ in range(retry_count): try: results = await self.call_procedure( procedure_name = procedure_name, procedure_args = procedure_args, ) except Exception as exc: exception = exc if exception is None: break await asyncio.sleep(backoff_seconds) backoff_seconds = backoff_seconds * backoff_multiplier # If the results are blank: if len(results) == 0: return { "status": results[0][0]["status"], "message": results[0][0].get("message", "please contact admin (E)"), "seconds": None, "data": {} } # Extract the very basic success or failure indicators: formatted_results = { "status": results[0][0]["status"], "message": results[0][0].get("message", "ok"), "seconds": 0.0, "data": {} } # Handle the remaining keys of the zeroth result set: for key, value in results[0][0].items(): if key not in formatted_results.keys(): formatted_results["data"][key] = value # Format for index in range(len(results)): if index > 0: formatted_results["data"][f"rs{index-1}"] = results[index] # Note down the time taken: formatted_results["seconds"] = time.perf_counter() - start_ts # Done here: if return_exception: return formatted_results, exception else: return formatted_results # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": async def test(connector): pass async def multi_test(count = 1): """ Test asynchronous behaviour. :return: None. """ # cred_json = { # "host": "del.ditscentre.in", # "user": "bicree", # "port": 3306, # "password": "9c3b2808a4aa281129d399fe09e69b53", # "database": "bicree" # } cred_json = { "host": "del.ditscentre.in", "user": "caOffice", "port": 3306, "password": "jstArchon", "database": "caOffice" } db_conn = AsyncMySQL( pool_size = 25, **cred_json ) await db_conn.connect() # result = await db_conn.call_procedure_and_get_json( # procedure_name = "login", # procedure_args = ( # "shree", # "shree", # "aiomysql", # "127.0.0.1" # ) # ) # result = await db_conn.call_procedure_and_get_json( # procedure_name = "listSummary", # procedure_args = ("bd7a6e53-1345-11ef-940c-0cc47a84a0bb",) # ) result = await db_conn.call_procedure_and_get_json( procedure_name = "campaign_activity_report", procedure_args = (10000000,) ) print("RESULT:", json.to_string(result, default = str)) start_time = time.time() asyncio.run(multi_test(count = 1)) print(f"ASYNC HITS DONE IN: {time.time() - start_time} seconds.")