""" AUTHOR: Khushal P Soonderji DATE: Tuesday, 22nd Oct., 2024 OBJECTIVE: To provide an easy way to create models to handle documents for Bicree. This is the base model for this microservice. It will define the structure for all other models that will be used in this particular microservice. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # My utils: from utils_v2.database.async_mysql_v2 import AsyncMySQL from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache from utils_v2.api.codes import StatusCodes # For asynchronous activities: import asyncio # For debugging: from icecream import IceCreamDebugger # To work with datatypes: from typing import List # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class BaseModel: PREVIEW_LENGTH = 250 def __init__( self, cache = None, alert_url = None, http_client = None, debug = True, debug_prefix = "Model | ", debug_only_errors = True ): """ This is the base model. :param cache: The object to use for caching results from database calls. :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 caching utility: self._cache = cache # For sending alerts: self._alert_url = alert_url self._http_client = http_client # Prepare the debugging utility: self._debug_prefix = debug_prefix self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) if not debug: self._printer.disable() self._debug_only_errors = debug_only_errors # A semaphore for activities that must absolutely be done one at a time: self.__exclusive_semaphore = asyncio.Semaphore(1) # A simple debugging output: self._printer("Model initialized.") def enable_terminal_print(self): self._printer.enable() def disable_terminal_print(self): self._printer.disable() def debug_only_errors(self): self._debug_only_errors = True def debug_everything(self): self._debug_only_errors = False async def send_alert( self, message: str, session_token = None, alert_type = "error" ): """ Sends out an alert (ideally through the tech module). This is meant to be used when some exception occurs, and you want to be informed before the client complains. :param message: The message to send out to the admins. :param session_token: The session token of the user (optional) so that the alert message can display the name of the user who faced the trouble. :param alert_type: The type of alert to throw ("error", "warning", or "info"). :return: None. """ if self._http_client is not None and self._alert_url is not None: response = await self._http_client.post( url = self._alert_url, json = { "sessionToken": session_token, "message": message, "type": alert_type } ) async def call_cached_procedure( self, cache: AsyncRedisCache, cache_key: str, cache_expiry: int, db_conn: AsyncMySQL, proc_name: str, proc_args: tuple, retry_count: int = 1, backoff_seconds: float = 0.5, backoff_multiplier: float = 1.1, session_token: str = None ): """ Calls a stored procedure and returns the response as a JSON-like object (dict or list). :param cache: The caching object to use to set the session in cache memory. :param cache_key: The string to use as the key when caching the response. :param cache_expiry: The no. of seconds after which this information will be deleted from the cache. :param db_conn: The connection instance to use to call the procedure. :param proc_name: The name of the stored procedure that must be called. :param proc_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 session_token: A session token to share with the tech module when alerts need to be sent out for any occurrence of exceptions. If this is passed, the tech module will be able to tell you which user faced the issue. :return: The response from the stored procedure. """ # check for the data in cache: data = await cache.get(cache_key) # If the data isn't in the cache, call the procedure: if data is None: # Make the database call: data = await self.call_procedure( db_conn = db_conn, proc_name = proc_name, proc_args = proc_args, retry_count = retry_count, backoff_seconds = backoff_seconds, backoff_multiplier = backoff_multiplier, session_token = session_token ) # If the database call succeeded, cache the response: if isinstance(data, dict) and data["status"] == 1: await cache.set(key = cache_key, value = data, expiry = cache_expiry) # Done here: return data async def call_procedure( self, db_conn: AsyncMySQL, proc_name: str, proc_args: tuple, retry_count: int = 1, backoff_seconds: float = 0.5, backoff_multiplier: float = 1.1, session_token: str = None ): """ Calls a stored procedure and returns the response as a JSON-like object (dict or list). :param db_conn: The connection instance to use to call the procedure. :param proc_name: The name of the stored procedure that must be called. :param proc_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 session_token: A session token to share with the tech module when alerts need to be sent out for any occurrence of exceptions. If this is passed, the tech module will be able to tell you which user faced the issue. :return: The response from the stored procedure. """ # Call the stored procedure: db_json, exception = await db_conn.call_procedure_and_get_json( proc_name, proc_args, retry_count = retry_count, backoff_seconds = backoff_seconds, backoff_multiplier = backoff_multiplier, return_exception = True ) # Understand the response: success = True if db_json["status"] == 1 else False message = db_json.get("message") # Debugging print: if not success or not self._debug_only_errors: self._printer(proc_name, proc_args, success, exception, message) # Send an alert out on exceptions: if exception is not None: # Format the message in Markdown format: exception_string = str(exception).replace("`", "'") formatted_message = f"*Module:*\n`{self._debug_prefix}`\n\n" formatted_message += f"*Proc:*\n`{proc_name}`\n\n" formatted_message += f"*Args:*\n`({', '.join([str(_) for _ in proc_args])})`\n\n" formatted_message += f"*Arg-Types:*\n`({', '.join([type(_).__name__ for _ in proc_args])})`\n\n" formatted_message += f"*Message:*\n`{message}`\n\n" formatted_message += f"*Success:*\n`{success}`\n\n" formatted_message += f"*Exception:*\n`{exception_string}`\n\n" # Send the alert: await self.send_alert(formatted_message, session_token = session_token) # Return the response: db_json["status_code"] = StatusCodes.OK if success else StatusCodes.FAILED return db_json async def execute_one( self, db_conn: AsyncMySQL, query: str, session_token: str = None ): """ Runs one query and sends an alert if that fails. :param db_conn: The connection to use to run the query. :param query: The query to run. :param session_token: A session token to share with the tech module when alerts need to be sent out for any occurrence of exceptions. If this is passed, the tech module will be able to tell you which user faced the issue. :return: The response from the database. """ # Run the query: rows_affected, db_response, exception = await db_conn.execute_one(query = query, return_exception = True) # Send an alert out on exceptions: if exception is not None: # Created needed previews: query_preview = query if len(query) <= self.PREVIEW_LENGTH else query[:self.PREVIEW_LENGTH] + "..." # Format the message in Markdown format: formatted_message = f"*Module:*\n`{self._debug_prefix}`\n\n" formatted_message += f"*Query:*\n`{query_preview}`\n\n" formatted_message += f"*Rows Affected:*\n`{rows_affected}`\n\n" formatted_message += f"*DB Response:*\n`{db_response}`\n\n" formatted_message += f"*Exception:*\n`{exception}`\n\n" # Send the alert: await self.send_alert(formatted_message, session_token = session_token) # Return the response: return rows_affected, db_response async def execute_many( self, db_conn: AsyncMySQL, query: str, data: List[tuple], session_token: str = None ): """ Runs many queries and sends an alert if that fails. :param db_conn: The connection to use to run the query. :param query: The query to run. :param data: The data to feed into the query. :param session_token: A session token to share with the tech module when alerts need to be sent out for any occurrence of exceptions. If this is passed, the tech module will be able to tell you which user faced the issue. :return: The response from the database. """ # Run the query: rows_affected, db_response, exception = await db_conn.execute_many( query = query, data = data, return_exception = True ) # Send an alert out on exceptions: if exception is not None: # Created needed previews: query_preview = query if len(query) <= self.PREVIEW_LENGTH else query[:self.PREVIEW_LENGTH] + "..." data_preview = str(data) if len(data_preview) > self.PREVIEW_LENGTH: data_preview = data_preview[:self.PREVIEW_LENGTH] + "..." # Format the message in Markdown format: formatted_message = f"*Module:*\n`{self._debug_prefix}`\n\n" formatted_message += f"*Query:*\n`{query_preview}`\n\n" formatted_message += f"*Data:*\n`{data_preview}`\n\n" formatted_message += f"*Rows Affected:*\n`{rows_affected}`\n\n" formatted_message += f"*DB Response:*\n`{db_response}`\n\n" formatted_message += f"*Exception:*\n`{exception}`\n\n" # Send the alert: await self.send_alert(formatted_message, session_token = session_token) # Return the response: return rows_affected, db_response # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass