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

Merge commit 'a4cb0e4ac1b4cfcca869ea87abe3ff126f23caf2' as 'utils_v2'

Browse Source
This commit is contained in:
khushal
2024-11-28 14:38:00 +05:30
parent bab6b34c7b a4cb0e4ac1
commit 90d5f99d26
108 changed files with 19627 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/database/__init__.py
View File
utils_v2/database/__pycache__/__init__.cpython-310.pyc
BIN
View File
Binary file not shown.
utils_v2/database/__pycache__/async_mongo_v2.cpython-310.pyc
BIN
View File
Binary file not shown.
utils_v2/database/__pycache__/async_mysql_v2.cpython-310.pyc
BIN
View File
Binary file not shown.
utils_v2/database/async_firebase.py
+504
View File
@@ -0,0 +1,504 @@
"""
AUTHOR:
Khushal P Soonderji
Bhushan Thakkar
DATE:
Friday, 19th April, 2024
OBJECTIVE:
To provide an easy interface to work with Firebase.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
# ---
import sys
sys.path.append(".")
sys.path.append("..")
# For async operations:
# ---
import asyncio
# For system-level activities:
# ---
import os
# My async utils:
# ---
import async_json_utils
# Firebase:
# ---
import firebase_admin
import firebase_admin.firestore_async
import firebase_admin.auth
import firebase_admin.db
# For debugging and logging:
# ---
from icecream import IceCreamDebugger
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
class AsyncFirebase:
def __init__(
self,
credentials_json_path,
app_name,
max_connections = 5,
debug = True
):
# Debugging print:
# ---
self.__printer = IceCreamDebugger(prefix = f"FBase ({app_name[:8]}) | ", includeContext = True)
if not debug: self.__printer.disable()
alert = f"Firebase session starting with max. {max_connections} connections."
self.__printer(alert)
# Initialize the instance:
# ---
self.__semaphore = asyncio.Semaphore(max_connections)
credentials = firebase_admin.credentials.Certificate(credentials_json_path)
self.__firebase_app = firebase_admin.initialize_app(credentials, name = app_name)
self.__firestore = firebase_admin.firestore_async.client(self.__firebase_app)
def __del__(self):
firebase_admin.delete_app(self.__firebase_app)
self.__firestore.close()
alert = "Firebase session ended."
self.__printer(alert)
def __user_to_json(self, firebase_user):
user_json = {
"uid": firebase_user.uid,
"email": firebase_user.email,
"emailVerified": firebase_user.email_verified,
"displayName": firebase_user.display_name,
"phoneNo": firebase_user.phone_number,
"photoUrl": firebase_user.photo_url,
"customClaims": firebase_user.custom_claims,
"disabled": firebase_user.disabled,
"providerId": firebase_user.provider_id,
"providerData": firebase_user.provider_data,
"tenantId": firebase_user.tenant_id
}
return user_json
async def create_custom_token(self, uid):
async with self.__semaphore:
try:
custom_token = firebase_admin.auth.create_custom_token(
uid = uid,
app = self.__firebase_app
).decode("utf-8")
return custom_token
except Exception as exception:
self.__printer(exception)
return None
async def authenticate_token(self, token):
"""
To authenticate the given session token.
:param token: The session token generated by Firebase on a successful sign-in.
:return: Either the retrieved user information or a blank dictionary.
"""
try:
user_info = firebase_admin.auth.verify_id_token(token, app = self.__firebase_app)
return user_info
except Exception as exception:
self.__printer(exception)
return {}
async def create_user(
self,
uid,
display_name = None,
password = None,
email = None,
phone_number = None
):
"""
Create a new user.
:param uid: The id to identify the user by.
:return: True or False based on the success of the operation.
"""
async with self.__semaphore:
try:
firebase_admin.auth.create_user(
uid = uid,
app = self.__firebase_app
)
return True
except firebase_admin.auth.UidAlreadyExistsError as excp:
return True
except Exception as exception:
self.__printer(exception)
return False
async def delete_user(self, uid):
async with self.__semaphore:
try:
firebase_admin.auth.delete_user(
uid = uid,
app = self.__firebase_app
)
return True
except firebase_admin.auth.UserNotFoundError as excp:
return True
except Exception as exception:
self.__printer(exception)
return False
async def get_user(self, uid):
async with self.__semaphore:
try:
user = firebase_admin.auth.get_user(
uid = uid,
app = self.__firebase_app
)
return self.__user_to_json(user)
except Exception as exception:
self.__printer(exception)
return None
async def list_users(self, users_per_page, page_token = None):
"""
To get a list of users. Firebase has a limit of 1000 per call of this API. So we use page-tokens to fetch next
pages of users.
:param users_per_page: How many users you want to list in this API call (Max. 1,000).
:param page_token: To be used in case of pagination.
:return: The list of users and the page token to be used for the next call. Will be None in case of failure.
"""
async with self.__semaphore:
try:
users = firebase_admin.auth.list_users(
page_token = page_token,
max_results = min(users_per_page, 1000),
app = self.__firebase_app
)
next_page_token = users.next_page_token if users.has_next_page else None
users = users.users
users_json = [self.__user_to_json(user) for user in users]
return users_json, next_page_token
except Exception as exception:
self.__printer(exception)
return None, None
async def create_document(self, collection_path, document_name, document_data = None):
"""
To create a new document with the specified data in an existing collection.
:param collection_path: The path of the collection (can be a sub-collection).
:param document_name: The name of the document you want to create.
:param document_data: The data that you want to populate in the document.
:return: True or False based on the success of the operation.
"""
async with self.__semaphore:
try:
snapshot = await self.__firestore.collection(
collection_path
).add(
document_id = document_name,
document_data = document_data or {}
)
return True
except Exception as exception:
self.__printer(exception)
return False
async def delete_document(self, path):
"""
To delete a document.
:param path: The path of the document.
:return: True or False based on the success of the operation.
"""
async with self.__semaphore:
try:
snapshot = await self.__firestore.document(path).delete()
return True
except Exception as excp:
print("FIRESTORE DOCUMENT DELETION EXCEPTION:", excp)
return False
async def set_document(self, path, data):
"""
To overwrite the data in a document.
:param path: The path of the document.
:param data: The data you want to update as a dictionary.
:return: True or False based on the success of the operation.
"""
async with self.__semaphore:
try:
snapshot = await self.__firestore.document(path).set(data)
return True
except Exception as exception:
self.__printer(exception)
return False
async def update_document(self, path, data):
"""
To update the data in a document. Provide only the fields that you want to update.
:param path: The path of the document.
:param data: The data you want to update as a dictionary.
:return: True or False based on the success of the operation.
"""
async with self.__semaphore:
try:
snapshot = await self.__firestore.document(path).update(data)
return True
except Exception as exception:
self.__printer(exception)
return False
async def get_document(self, path):
"""
To get the data in a document.
:param path: The path of the document.
:return: The dictionary of data as found in the path specified or None if the operation failed.
"""
async with self.__semaphore:
try:
snapshot = await self.__firestore.document(path).get()
return snapshot.to_dict()
except Exception as exception:
self.__printer(exception)
return None
async def get_document_fields(self, path, fields):
"""
To get only specific fields (keys) in a document. Like how projections are used in Mongo.
:param path: The path of the document.
:param fields: The list of fields (keys) of the document that you want.
:return: The dictionary of data as found in the path specified.
"""
async with self.__semaphore:
try:
if type(fields) is not list: fields = [fields]
snapshot = await self.__firestore.document(path).get(fields)
return snapshot.to_dict()
except Exception as exception:
self.__printer(exception)
return None
async def create_collection(self, collection_name, document_name, document_data):
"""
Creates a new collection in the root of the database. Note that Firebase doesn't allow creating new empty
collections, so we must add one first document in it.
:param collection_name: The name of the collection you want to create.
:param document_name: The name of the first document you want to put in the collection.
:param document_data: The data that you want to put in the first document of the new collection.
:return: True or False based on the success of the operation.
"""
async with self.__semaphore:
try:
snapshot = await self.__firestore.collection(
collection_name
).add(
document_id = document_name,
document_data = document_data
)
return True
except Exception as exception:
self.__printer(exception)
return False
async def create_sub_collection(
self,
document_path,
sub_collection_name,
sub_document_name = None,
sub_document_data = None
):
"""
Firebase allows you to create collections inside documents. This method is built for that. Note that Firebase
doesn't allow creating new empty collections, so we must add one first document in it.
:param document_path: The path of the document in which you want to create a new collection.
:param sub_collection_name: The name of the collection you want to create.
:param sub_document_name: The name of the first document you want to put in the collection.
:param sub_document_data: The data that you want to put in the first document of the new collection.
:return: True or False based on the success of the operation.
"""
async with self.__semaphore:
try:
snapshot = await self.__firestore.document(
document_path
).collection(
sub_collection_name
).add(
document_id = sub_document_name,
document_data = sub_document_data or {}
)
return True
except Exception as exception:
self.__printer(exception)
return False
async def get_collection(self, path):
"""
Get a whole collection's data.
:param path: The path of the collection.
:return: The collection.
"""
async with self.__semaphore:
try:
snapshots = await self.__firestore.collection(path).get()
return {snapshot.id: snapshot.to_dict() for snapshot in snapshots}
except Exception as exception:
self.__printer(exception)
return None
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
import time
from shared import constants
user_data = {
"firebaseUid": None,
"roles": "ClientAdmin",
"sessionToken": "siddhesh_20240419"
}
async def test():
my_firebase = AsyncFirebase(
f"{constants.PROJECT_DIRECTORY}/utils/cred/firebase_certs.json",
"myFire",
max_connections = 5
)
print("\n\n---\n\n")
results = await my_firebase.get_document("testCollection/testDoc")
print(async_json_utils.to_json_string(results))
print("\n\n---\n\n")
results = await my_firebase.get_document_fields("testCollection/testDoc", ["sampleMap"])
print(async_json_utils.to_json_string(results))
# await my_firebase.create_sub_collection(
# "myDeepCollection/deepDocId",
# "subCollection3",
# "subDoc",
# {"sub_key": "sub_val"}
# )
# await my_firebase.create_collection(
# "rootCollection",
# "subDoc",
# {"sub_key": "sub_val"}
# )
# await my_firebase.delete_document("activeSessions/9d402f1502dfd55a4326fa7fc8e6cb7d")
# print(async_json_utils.to_json_string(await my_firebase.get_collection("myDeepCollection")))
start_time = time.time()
asyncio.run(test())
print(f"FINISHED IN {time.time() - start_time} SECONDS.")
utils_v2/database/async_mongo.py
+843
View File
@@ -0,0 +1,843 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Tuesday, 28th May, 2024
OBJECTIVE:
To have one central place from where all async database connectivity happens.
REFERENCES:
01. https://motor.readthedocs.io/en/stable/
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# For async behaviour:
import asyncio
# For datetime handling:
import pytz
import datetime
# MongoDB:
from motor.motor_asyncio import AsyncIOMotorClient
from bson.objectid import ObjectId
from bson.json_util import dumps, loads
# For debugging:
from icecream import IceCreamDebugger
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** EXCEPTIONS ***
# ***** ****
# *****************************************************************************************************************
class MongoFindException(Exception):
def __init__(self, hint = None, origin = None):
self.__hint = hint
self.__origin = origin
def __str__(self):
message = "mongo find operation failed"
if self.__hint is not None: message = f"{message} ({self.__hint})"
if self.__origin is not None: message = f"{self.__origin} --> {message}"
return message
# ---------------------------------------------------------------------------------------------------------------------
class MongoInsertException(Exception):
def __init__(self, hint = None, origin = None):
self.__hint = hint
self.__origin = origin
def __str__(self):
message = "mongo insert operation failed"
if self.__hint is not None: message = f"{message} ({self.__hint})"
if self.__origin is not None: message = f"{self.__origin} --> {message}"
return message
# ---------------------------------------------------------------------------------------------------------------------
class MongoUpdateException(Exception):
def __init__(self, hint = None, origin = None):
self.__hint = hint
self.__origin = origin
def __str__(self):
message = "mongo update operation failed"
if self.__hint is not None: message = f"{message} ({self.__hint})"
if self.__origin is not None: message = f"{self.__origin} --> {message}"
return message
# ---------------------------------------------------------------------------------------------------------------------
class MongoReplaceException(Exception):
def __init__(self, hint = None, origin = None):
self.__hint = hint
self.__origin = origin
def __str__(self):
message = "mongo replace operation failed"
if self.__hint is not None: message = f"{message} ({self.__hint})"
if self.__origin is not None: message = f"{self.__origin} --> {message}"
return message
# ---------------------------------------------------------------------------------------------------------------------
class MongoDeleteException(Exception):
def __init__(self, hint = None, origin = None):
self.__hint = hint
self.__origin = origin
def __str__(self):
message = "mongo delete operation failed"
if self.__hint is not None: message = f"{message} ({self.__hint})"
if self.__origin is not None: message = f"{self.__origin} --> {message}"
return message
# ---------------------------------------------------------------------------------------------------------------------
class MongoException(Exception):
def __init__(self, hint = None, origin = None):
self.__hint = hint
self.__origin = origin
def __str__(self):
message = "mongo operation failed"
if self.__hint is not None: message = f"{message} ({self.__hint})"
if self.__origin is not None: message = f"{self.__origin} --> {message}"
return message
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AsyncMongo:
__db = None
__db_name = None
__client = None
def __init__(
self,
database_name = "myDb",
max_connections = 5,
debug = True,
debug_only_errors = True,
host = "localhost",
port = 27017,
connection_string = None
):
# Basic config:
self.__max_connections = max_connections
self.__db_name = database_name
self.__host = host,
self.__port = port
self.__connection_string = connection_string
# For debugging:
self.__debug_only_errors = debug_only_errors
self.__printer = IceCreamDebugger(prefix = f"Mongo ({self.__db_name}) | ", includeContext = True)
if not debug: self.__printer.disable()
# Rate/access control:
self.__exclusive_semaphore = asyncio.Semaphore(1)
def enable_debug(self):
self.__printer.enable()
def disable_debug(self):
self.__printer.disable()
@staticmethod
def generate_id():
"""
Just generates an '_id' in MongoDB style.
:return: The '_id' in MongoDB style.
"""
return str(ObjectId())
async def connect(self):
try:
if self.__connection_string is None:
self.__client = AsyncIOMotorClient(
self.__host,
self.__port,
maxPoolSize = self.__max_connections,
minPoolSize = self.__max_connections
)
else:
self.__client = AsyncIOMotorClient(
self.__connection_string,
maxPoolSize = self.__max_connections,
minPoolSize = self.__max_connections
)
self.__db = self.__client.get_database(self.__db_name)
if not self.__debug_only_errors:
server_info = await self.__client.server_info()
self.__printer(server_info)
except Exception as exception: self.__printer(exception)
async def ensure_connection(self):
if self.__db is None:
async with self.__exclusive_semaphore:
await self.connect()
@property
async def client(self):
await self.ensure_connection()
return self.__client
@staticmethod
def dict_to_dot_notation(input_dict, parent_key = "", separator = "."):
items = []
for k, v in input_dict.items():
new_key = f"{parent_key}{separator}{k}" if parent_key else k
if isinstance(v, dict) and v:
items.extend(AsyncMongo.dict_to_dot_notation(v, new_key, separator = separator).items())
else:
items.append((new_key, v))
return dict(items)
@staticmethod
def normalize_date_time(document):
"""
MongoDB doesn't support timezones. A good strategy would be to convert everything to UTC format and store it.
This method does exactly that. Any datetime object is converted to UTC timezone. If the datetime object was
timezone naive, UTC timezone will be applied to it without changing the time value.
:param document: The document that you want to normalize the date-time in.
:return: The document with normalized datetime.
"""
if isinstance(document, datetime.datetime):
utc_tz = pytz.timezone("UTC")
if document.tzinfo is None: document = utc_tz.localize(document)
else: document = document.astimezone(utc_tz)
if type(document) is list:
document = [AsyncMongo.normalize_date_time(item) for item in document]
if type(document) is dict:
document = {
AsyncMongo.normalize_date_time(k): AsyncMongo.normalize_date_time(v)
for k, v in document.items()
}
return document
@staticmethod
def __from_json_string(json_data):
return loads(json_data)
@staticmethod
def __to_json_string(python_data, indent = 4, default = None):
return dumps(python_data, indent = indent, default = default)
async def list_indexes(
self,
collection,
session = None,
raise_exception = False
):
"""
Lists out the indexes of a collection.
:param collection: The collection whose indexes you want to list out,
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: The list of indexes or None if the action fails.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
indexes = None
# Try to list the indexes:
try:
responses = await self.__db[collection].list_indexes(session = session).to_list(None)
indexes = [{key: value for key, value in response.items()} for response in responses]
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not indexes: raise MongoException(hint = "list indexes")
return indexes
async def create_index(
self,
collection,
keys,
options = None,
session = None,
raise_exception = False
):
"""
Creates an index on a collection.
:param collection: The collection to create the index on.
:param keys: The keys (and sorting) to implement the index on.
:param options: Additional config.
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: True or False based on the success of the execution.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
success = False
# Try to make the insertion:
try:
options = options or {}
keys = [(k, v) for k, v in keys.items()]
response = await self.__db[collection].create_index(keys, session = session, **options)
if response: success = True
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not success: raise MongoException(hint = "create index")
return success
async def insert_one(
self,
collection,
document,
session = None,
raise_exception = False
):
"""
Insert data into a collection.
:param collection: The collection you want to feed the data into.
:param document: The data to be stored.
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: The id of the inserted data, or null if the action fails.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
inserted_id = None
# Try to make the insertion:
try:
response = await self.__db[collection].insert_one(document.copy(), session = session)
inserted_id = response.inserted_id
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not inserted_id: raise MongoInsertException(hint = f"{collection}")
return inserted_id
async def insert_many(
self,
collection,
documents,
session = None,
raise_exception = False
):
"""
Insert a lot of data into a collection.
:param collection: The collection you want to feed the data into.
:param documents: The data to be stored.
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: The id of the inserted data, or null if the action fails.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
inserted_ids = []
# Try to make the insertion:
try:
response = await self.__db[collection].insert_many(documents, session = session)
inserted_ids = response.inserted_ids
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not inserted_ids: raise MongoInsertException(hint = f"{collection}")
return inserted_ids
async def update_one(
self,
collection,
filter_json = None,
update_json = None,
upsert = False,
session = None,
raise_exception = False
):
"""
Update one document.
:param collection: The collection you want to update.
:param filter_json: The selection criteria to locate the document to update.
:param update_json: The values you want to update.
:param upsert: If you want to insert if the document doesn't already exist.
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: True or False based on the success of the operation.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
success = False
# Try to make the insertion:
try:
response = await self.__db[collection].update_one(
filter_json,
update_json,
upsert = upsert,
session = session
)
success = False if response.modified_count == 0 else True
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not success: raise MongoUpdateException(hint = f"{collection}")
return success
async def update_many(
self,
collection,
filter_json = None,
update_json = None,
upsert = False,
session = None,
raise_exception = False
):
"""
Update many documents.
:param collection: The collection you want to update.
:param filter_json: The selection criteria to locate the document to update.
:param update_json: The values you want to update.
:param upsert: If you want to insert if the document doesn't already exist.
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: True or False based on the success of the operation.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
update_count = 0
# Try to make the insertion:
try:
response = await self.__db[collection].update_many(
filter_json,
update_json,
upsert = upsert,
session = session
)
update_count = response.modified_count
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not update_count: raise MongoUpdateException(hint = f"{collection}")
return update_count
async def count(
self,
collection,
filter_json = None
):
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
count = 0
# try to query the data:
try:
if filter_json is None: filter_json = {}
count = await self.__db[collection].count_documents(filter_json)
except Exception as exception: self.__printer(exception)
# Done here:
return count
async def bulk_write(
self,
collection,
requests,
session = None,
raise_exception = False
):
"""
To perform various individual operations in one go. You will have to import individual actions like "UpdateOne"
and "InsertMany" from PyMongo and pass them as an array of requests (operations) to this method.
:param collection: The collection you want to run the requests on.
:param requests: The array of requests (operations) to be performed.
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return:
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
count = 0
# try to query the data:
try:
response = await self.__db[collection].bulk_write(requests, session = session)
count = response.modified_count + response.inserted_count + response.upserted_count + response.deleted_count
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not count: raise MongoException(hint = "bulk write")
return count
async def find_many(
self,
collection,
filter_json,
projections = None,
skip = 0,
limit = None,
sort = None,
session = None,
as_json_string = False,
indent = 4,
raise_exception = False
):
"""
Finds one or more records that match the given conditions.
:param collection: The name of the collection to perform the search in.
:param filter_json: The filter criteria.
:param projections: What parts of the matching data you want to fetch.
:param skip: The no. of records to skip before picking next ones. Needed for pagination.
:param limit: The max. no. of records you want to fetch.
:param sort: The sorting rules to apply.
:param session: The session if you need to do this in a transaction.
:param as_json_string: Whether you want it as a JSON string or a Python dict/list.
:param indent: The indentation to use if you want it as a JSON string.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: The array of matching records or null if there was an exception.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
results = None
# Assume defaults:
if sort is None: sort = {"_id": -1}
if limit is None: limit = 10
# try to query the data:
try:
results = await self.__db[collection].find(
filter_json,
projections,
session = session
).sort(sort).skip(skip).limit(limit).to_list(None)
if as_json_string: results = self.__to_json_string(results, indent = indent, default = str)
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not results: raise MongoFindException(hint = f"{collection}")
return results
async def find_one(
self,
collection,
filter_json,
projections = None,
session = None,
as_json_string = False,
indent = 4,
raise_exception = False
):
"""
Finds one record that matches the given conditions.
:param collection: The name of the collection to perform the search in.
:param filter_json: The filter criteria.
:param projections: What parts of the matching data you want to fetch.
:param session: The session if you need to do this in a transaction.
:param as_json_string: Whether you want it as a JSON string or a Python dict/list.
:param indent: The indentation to use if you want it as a JSON string.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: The array of matching records or null if there was an exception.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
results = None
# try to query the data:
try:
results = await self.__db[collection].find_one(filter_json, projections, session = session)
if as_json_string: results = self.__to_json_string(results, indent = indent, default = str)
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not results: raise MongoFindException(hint = f"{collection}")
return results
async def replace_one(
self,
collection,
filter_json,
replacement_json,
upsert = False,
session = None,
raise_exception = False
):
"""
To delete one document from a collection.
:param collection: The collection from which you want to delete many records.
:param filter_json: The filter criteria.
:param replacement_json: The data to put in place of the existing document.
:param upsert: If you want to insert if the document doesn't already exist.
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: True or False based on the success of the operation.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
success = False
# try to query the data:
try:
result = await self.__db[collection].replace_one(
filter_json,
replacement_json,
upsert = upsert,
session = session
)
if result.modified_count or result.upserted_id: success = True
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not success: raise MongoReplaceException(hint = f"{collection}")
return success
async def delete_one(
self,
collection,
filter_json,
session = None,
raise_exception = False
):
"""
To delete one document from a collection.
:param collection: The collection from which you want to delete many records.
:param filter_json: The filter criteria.
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: True or False based on the success of the operation.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
count = 0
# try to query the data:
try:
result = await self.__db[collection].delete_one(filter_json, session = session)
count = result.deleted_count
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not count: raise MongoDeleteException(hint = f"{collection}")
return count
async def delete_many(
self,
collection,
filter_json,
session = None,
raise_exception = False
):
"""
To delete many documents from a collection.
WARNING: sending {} in the filter would mean deleting ALL the documents.
:param collection: The collection from which you want to delete many records.
:param filter_json: The filter criteria.
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: True or False based on the success of the operation.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
count = 0
# try to query the data:
try:
result = await self.__db[collection].delete_many(filter_json, session = session)
count = result.deleted_count
except Exception as exception:
self.__printer(exception)
# Check results and return:
if raise_exception and not count: raise MongoDeleteException(hint = f"{collection}")
return count
async def aggregate(
self,
collection,
pipeline,
limit = None,
raise_exception = False
):
"""
Perform an advance query on the data.
:param collection: The collection to perform the query on.
:param pipeline: The pipeline of actions to take. Must be a list.
:param limit: The max. no. of records to retrieve.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return: The array of matching records or null if there was an exception.
"""
# Ensure you are connected:
await self.ensure_connection()
# Assume failure:
results = None
# try to perform the aggregation action:
try: results = await self.__db[collection].aggregate(pipeline).to_list(limit)
except Exception as exception: self.__printer(exception)
# Check results and return:
if raise_exception and not results: raise MongoException(hint = "aggregation")
return results
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
utils_v2/database/async_mongo_storage.py
+358
View File
@@ -0,0 +1,358 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Tuesday, 28th May, 2024
OBJECTIVE:
To have one central place from where all async database connectivity happens.
REFERENCES:
01. https://motor.readthedocs.io/en/stable/
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
# ---
import sys
sys.path.append(".")
sys.path.append("..")
# For system-level activity:
import io
# For async behaviour:
import asyncio
# MongoDB for File Storage:
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorGridFSBucket
from bson.objectid import ObjectId
# For debugging:
from icecream import IceCreamDebugger
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AsyncMongoStorage:
__db = None
__db_name = None
__client = None
__store = None
def __init__(
self,
connection_string = None,
max_connections = 5,
host_name = "localhost",
port = 27017,
database_name = "fileStore",
debug = True,
debug_prefix = "GridFS (M) | ",
debug_only_errors = True
):
# Database Initialization:
self.__host_name = host_name
self.__port = port
self.__db_name = database_name
self.__connection_string = connection_string
self.__max_connections = max_connections
# Debugging:
self.__debug_only_errors = debug_only_errors
self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
if not debug: self.__printer.disable()
@staticmethod
def generate_id():
"""
Just generates an '_id' in MongoDB style.
:return: The '_id' in MongoDB style.
"""
return str(ObjectId())
async def connect(self):
"""
Initialize the database connection.
:return: Nothing.
"""
if self.__connection_string is None:
self.__client = AsyncIOMotorClient(
self.__host_name,
self.__port,
maxPoolSize = self.__max_connections,
minPoolSize = self.__max_connections
)
else:
self.__client = AsyncIOMotorClient(
self.__connection_string,
maxPoolSize = self.__max_connections,
minPoolSize = self.__max_connections
)
self.__db = self.__client.get_database(self.__db_name)
self.__store = AsyncIOMotorGridFSBucket(self.__db)
@property
def fs(self):
"""
To access the features that have not been wrapped in this reportlab directly.
This could include things like streaming files chunk-by-chunk.
:return: The file-store instance.
"""
return self.__store
async def write_from_memory(self, file_name, file_data, metadata_json = None):
"""
Save a file (from RAM) to Mongo. Suitable for smaller files (a few MBs max.).
:param file_name: The name of the file.
:param file_data: The data of the file (held in RAM).
:param metadata_json: A JSON of metadata information that can later be used to search files (RECOMMENDED).
:return: The file's id as a string (if it gets saved) or None.
"""
if self.__store is None: await self.connect()
file_data.seek(0)
file_size = file_data.__sizeof__()
file_id = None
try: file_id = await self.__store.upload_from_stream(file_name, file_data, metadata = metadata_json)
except Exception as exception: self.__printer(exception, file_name, file_size, file_id)
if not self.__debug_only_errors: self.__printer(file_name, file_size, file_id)
return str(file_id)
async def read_to_memory(self, file_identifier, by_id = True):
"""
To retrieve a file (in RAM) based on the provided identifier.
Suitable for smaller files (a few MBs max.).
:param file_identifier: Either the name or the "_id" of the file.
:param by_id: Set to True if you are fetching by the "_id" of the file.
:return: Either the file (in RAM) or None.
"""
if self.__store is None: await self.connect()
file_data = None
try:
if by_id: grid_out = await self.__store.open_download_stream(ObjectId(file_identifier))
else: grid_out = await self.__store.open_download_stream_by_name(file_identifier)
file_data = io.BytesIO(await grid_out.read())
file_data.seek(0)
except Exception as exception:
file_data = None
self.__printer(exception, file_identifier, by_id)
if not self.__debug_only_errors: self.__printer(file_identifier, by_id)
return file_data
async def delete_file_by_id(self, file_id):
"""
Tries to delete one file by the id.
:param file_id: The id of the file in the database.
:return: True or False based on the success of the operation.
"""
if self.__store is None: await self.connect()
deleted = False
try:
response = await self.__store.delete(file_id = ObjectId(file_id))
deleted = True
except Exception as exception:
self.__printer(exception, file_id, deleted)
return deleted
def __format_metadata_json(self, metadata_json):
"""
NOTE: ONLY USE WHEN SEARCHING FILES BY METADATA.
MongoDB expects dot-notation while searching for files by the metadata. We are making a function to search
files assuming that the conditions are to be applied to the metadata itself. So this function add the
dot-notation to the right places to conduct a successful search.
:param metadata_json: The JSON to format.
:return: The formatted JSON that has the right dot-notation.
"""
formatted_metadata_json = {}
for key, value in metadata_json.items():
if not key.startswith("$"): key = f"metadata.{key}"
else:
if type(value) is dict: value = self.__format_metadata_json(value)
if type(value) is list: value = [self.__format_metadata_json(item) for item in value]
formatted_metadata_json[key] = value
return formatted_metadata_json
async def find_file_by_metadata(self, metadata_json, limit = None, skip = None, sort = None):
"""
This method only lists the files that match the criteria mentioned in the metadata JSON.
:param metadata_json: The JSON that describes what you want to find.
:param limit: Max. no. of records to retrieve.
:param skip: No. of starting results to skip. Useful for pagination.
:param sort: The sorting conditions to follow.
:return: A list of (JSONs of) files that match the conditions. The list can be empty.
"""
if self.__store is None: await self.connect()
files_list = []
try:
limit = limit or 10
skip = skip or 0
sort = {"_id": -1} if not isinstance(sort, dict) else sort
formatted_metadata_json = self.__format_metadata_json(metadata_json)
return await self.__store.find(
formatted_metadata_json
).sort(sort).skip(skip).limit(limit).to_list(None)
except Exception as exception: self.__printer(exception, metadata_json, len(files_list))
if not self.__debug_only_errors: self.__printer(metadata_json, len(files_list))
return files_list
async def find_file_by_id(self, file_id):
"""
This method allows you to get the file's info from the id of the file.
:param file_id: The id that was assigned by Mongo during upload.
:return: The file's info or None if the file doesn't exist.
"""
if self.__store is None: await self.connect()
file_info = None
try:
formatted_metadata_json = {"_id": ObjectId(file_id)}
file_info = (await self.__store.find(formatted_metadata_json).to_list(1))[0]
except Exception as exception: self.__printer(exception, file_id, file_info)
return file_info
async def get_file_name(self, file_id):
"""
Returns the file name if the id of the file is known.
:param file_id: The id of the file as assigned by MongoDB when the file was stored.
:return: The file's name (if it exists), or None.
"""
# Ensure that we are connected:
if self.__store is None: await self.connect()
# Ensure that the input given is of 'ObjectId' type:
if type(file_id) is not ObjectId: file_id = ObjectId(str(file_id))
# Fetch and return the file name:
files_list = await self.__store.find(
{"_id": file_id},
{"filename": True}
).sort({"_id": -1}).limit(1).to_list(None)
try: file_name = files_list[0]["filename"]
except: file_name = None
return file_name
async def aggregate(
self,
collection,
pipeline,
limit = None
):
"""
Perform an advance query on the data.
:param collection: The collection to perform the query on.
:param pipeline: The pipeline of actions to take. Must be a list.
:param limit: The max. no. of records to retrieve.
:return: The array of matching records or null if there was an exception.
"""
if self.__store is None: await self.connect()
results = None
try: results = await self.__db[collection].aggregate(pipeline).to_list(limit)
except Exception as exception: self.__printer(exception)
return results
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
utils_v2/database/async_mongo_v2.py
+1736
View File
File diff suppressed because it is too large Load Diff
utils_v2/database/async_mysql_v2.py
+424
View File
@@ -0,0 +1,424 @@
"""
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
import decimal
# 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
# To work with datatypes:
from typing import List
# *****************************************************************************************************************
# ***** ****
# *** 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()
@staticmethod
def __parse_row(row):
"""
Converts from the custom objects of 'aiomysql' to types that are supported by Python.
:param row: The row from the result.
:return: The parsed row which will have types that are closer to being native to Python..
"""
parsed_row = []
for item in row:
if isinstance(item, decimal.Decimal): parsed_row.append(float(item))
else: parsed_row.append(item)
return parsed_row
async def fetch_all(self, cursor):
# Make a variable to hold all the result sets.
# Needed for when the procedure responds with many "tables":
all_result_sets = []
# Iterate over all result sets,
# and process them one-by-one:
while True:
# Process the current result set:
this_result_set = []
result = await cursor.fetchall()
if not cursor.description: break
columns = [desc[0] for desc in cursor.description]
for row in result: this_result_set.append(dict(zip(columns, self.__parse_row(row))))
all_result_sets.append(this_result_set)
# Move to the next set,
# or break out of the loop if all done:
if not await cursor.nextset(): break
# Done here:
return all_result_sets
async def call_procedure(self, procedure_name, procedure_args):
"""
To call stored procedures and retrieve all the responses.
:param procedure_name: The name of the stored procedure that must be called.
:param procedure_args: The args to be sent to the stored procedure.
:return: The raw result set as received from the database.
"""
# 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)
all_result_sets = await self.fetch_all(cursor)
# # Iterate over all result sets,
# # and process them one-by-one:
# while True:
# this_result_set = []
# result = await cursor.fetchall()
# if not cursor.description: break
# columns = [desc[0] for desc in cursor.description]
# for row in result: this_result_set.append(dict(zip(columns, self.__parse_row(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
):
"""
The method to call when you need to call a stored procedure and retrieve the response as a JSON-like object.
This is custom formatting based on the structure created by Mr. bhushan Thakkar in late April (2024).
:param procedure_name: The name of the stored procedure that must be called.
:param procedure_args: The args to be sent to the stored procedure.
:param retry_count: The max. number of times to try in case one or more attempts fail.
:param backoff_seconds: The time to wait before making the next attempt if the retry count is more than 1.
:param backoff_multiplier: The factor that dictates how much to modify the time delay by when waiting to retry.
:param return_exception: Whether, or not, you would like to return the exception object if something goes wrong.
:return: The formatted response and the exception (if asked for).
"""
# Note down the start time:
start_ts = time.time()
# 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:
formatted_results = {
"status": 0,
"message": "Please contact admin (NE)" if exception is None else "Please contact admin (E)",
"seconds": time.time() - start_ts,
"data": {}
}
if return_exception: return formatted_results, exception
else: return formatted_results
# 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.time() - start_ts
# Done here:
if return_exception: return formatted_results, exception
else: return formatted_results
async def execute_one(
self,
query: str,
commit: bool = True,
return_exception: bool = False
):
"""
Runs one command / query in SQL.
:param query: The query / command to run.
:param commit: Whether, or not, you would like to commit the execution.
:param return_exception: Whether, or not, you would like to return the exception from this function.
:return: Either just the result or the result and the exception.
"""
# Make sure we have a connection:
await self.ensure_connection()
# Start by assuming failure:
rows_affected = None
results = None
excp = None
try:
# Get a connection and execute the command:
async with self.__pool.acquire() as connection:
async with connection.cursor() as cursor:
rows_affected = await cursor.execute(query)
results = await self.fetch_all(cursor)
if commit: await connection.commit()
# SQL-specific errors:
except aiomysql.MySQLError as exception:
self.__printer("SQL Exception", exception)
excp = exception
# Other errors:
except Exception as exception:
self.__printer("Other Exception", exception)
excp = exception
# Done here:
if return_exception: return rows_affected, results, excp
else: return rows_affected, results
async def execute_many(
self,
query: str,
data: List[tuple],
commit: bool = True,
return_exception: bool = False
):
"""
Runs many commands / queries in SQL.
Consider the following example:
QUERY: "INSERT INTO pincodeMaster (pincode, city, state) VALUES (%s, %s, %s);"
DATA: [
('110001', 'New Delhi', 'Delhi'),
('500001', 'Hyderabad', 'Telangana'),
('600001', 'Chennai', 'Tamil Nadu')
]
:param query: The query / command to run.
:param data: The data to substitute into the query string.
:param commit: Whether, or not, you would like to commit the execution.
:param return_exception: Whether, or not, you would like to return the exception from this function.
:return: Either just the result or the result and the exception.
"""
# Make sure we have a connection:
await self.ensure_connection()
# Start by assuming failure:
rows_affected = None
results = None
excp = None
try:
# Get a connection and execute the command:
async with self.__pool.acquire() as connection:
async with connection.cursor() as cursor:
rows_affected = await cursor.executemany(query, data)
results = await self.fetch_all(cursor)
if commit: await connection.commit()
# SQL-specific errors:
except aiomysql.MySQLError as exception:
self.__printer("SQL Exception", exception)
excp = exception
# Other errors:
except Exception as exception:
self.__printer("Other Exception", exception)
excp = exception
# Done here:
if return_exception: return rows_affected, results, excp
else: return rows_affected, results
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass