khushalps/api_internal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Squashed 'utils_v2/' content from commit ef9630d

Browse Source 
git-subtree-dir: utils_v2
git-subtree-split: ef9630d728847764d4832ce8f3f956571901383a
This commit is contained in:
khushal
2024-10-01 10:33:59 +05:30
commit 8494a2c0c0
84 changed files with 13541 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database/async_mysql_v2.py
+311
View File
@@ -0,0 +1,311 @@
"""
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.")