312 lines
10 KiB
Python
312 lines
10 KiB
Python
"""
|
|
|
|
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).
|
|
: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
|
|
|
|
print("LEN:", len(results))
|
|
|
|
# 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:
|
|
print(f"{procedure_name}:")
|
|
print(json.to_string(formatted_results))
|
|
print("\n")
|
|
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"
|
|
}
|
|
|
|
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",)
|
|
)
|
|
|
|
|
|
start_time = time.time()
|
|
asyncio.run(multi_test(count = 1))
|
|
print(f"ASYNC HITS DONE IN: {time.time() - start_time} seconds.")
|
|
|