""" AUTHOR: Khushal P Soonderji DATE: Tuesday, 8th Oct., 2024 OBJECTIVE: This is the central location for the Quart module. We define the app here, and import and attach all blueprints here. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For system level activities: import gc import os import psutil # For using Quart: from quart import Quart, request, current_app from quart_cors import cors # Common: from shared import constants # My utils: from utils_v2.string import json from utils_v2.api import async_quart from utils_v2.date_time import date_time from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache from utils_v2.serialization.json_serializer import JSONSerializer from utils_v2.api.async_quart import ( set_api_version, read_input, log_request_to_mongo, should_not_be_under_maintenance, only_whitelisted_ips, limit_rate, validate_input, handle_cancelled_request ) # GMail-related utils: from utils_v2.goog.gmail.gmail_client import AsyncGMailClient, SCOPES_GMAIL_MAIL_MANAGEMENT # To make REST API calls: import httpx # For debugging: from icecream import IceCreamDebugger # All the blueprints: from api.blueprints.mail.oauth import mail_oauth_bp from api.blueprints.mail.callback import mail_callback_bp from api.blueprints.test.callback import test_callback_bp # All the helpers: from api.helpers.user import session # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Quart related: MODULE_BASE = constants.MODULE_NAME APP_VERSION = constants.APP_VERSION # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # The Quart app: app = Quart(__name__) app = cors(app) app.register_blueprint(mail_oauth_bp, url_prefix = f"/{MODULE_BASE}/mail") app.register_blueprint(mail_callback_bp, url_prefix = f"/{MODULE_BASE}/mail") app.register_blueprint(test_callback_bp, url_prefix = f"/{MODULE_BASE}/test") # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** def set_cpu_affinity(requested_cpus: list): """ Sets the affinity of the current process to certain CPUs so that performance is boosted. The main factors that contribute to gains are cache-locality, reduced context switching, and effective resource management. :param requested_cpus: The array of integers of which CPU cores are preferred. :return: None. """ # Get the number of available CPUs: num_cpus = psutil.cpu_count() # Wrap around logic for when a core has been request that doesn't exist on this machine. # This is useful in cases like developing on a local machine with just 4 cores, but your server has dozens of cores. valid_cpus = [cpu % num_cpus for cpu in requested_cpus] # Set the CPU affinity: psutil.Process().cpu_affinity(valid_cpus) # --------------------------------------------------------------------------------------------------------------------- @app.before_serving @set_api_version(api_version = APP_VERSION) @log_request_to_mongo( attr_name = "logs_mongo", project = constants.PROJECT_NAME, log_type = constants.MODULE_NAME, operation = "apiStart", log_input = True, log_output = True ) async def app_startup(**kwargs): """ To initialize the variables that you would like to use in this module. WARNING: ALL VARIABLES WILL BE INITIALIZED 'n' NUMBER OF TIMES, WHERE 'n' IS THE COUNT OF WORKERS DEPLOYED. SO, IF YOU WANT TO CONNECT TO A DATABASE AND YOU ALLOW A POOL-SIZE OF 10 AND IF YOU DEPLOY 4 WORKERS, YOU WILL END UP WITH 40 CONNECTIONS TO THE DATABASE. :return: None. """ # Safe-halt mechanism for upgrades (for a single-worker run): current_app.is_under_maintenance = False # Debugging: enable_debugging = True if os.environ["DEBUG"].strip().lower() == "true" else False current_app.printer = IceCreamDebugger(prefix = f"{MODULE_BASE} (Q) | ", includeContext = True) if not enable_debugging: current_app.printer.disable() current_app.printer("initializing worker...") # Make an instance of an HTTP client to use to make API calls: current_app.http_client = httpx.AsyncClient( limits = httpx.Limits( max_connections = 100, # ............ Maximum number of connections allowed in the pool. max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive. ), timeout = httpx.Timeout( connect = 2.5, # ... Shorter connection timeout. read = 2.5, # ...... Like what EasyEcom gives. write = 10.0, # .... Time to wait for sending data. pool = 120.0 # ..... Time to wait for a free connection from the pool. ) ) # Get the script credentials and data: script_id = os.environ["SCRIPT_ID"] response = await current_app.http_client.get( url = r"https://nexcom.ditscentre.in/internal/cred/get", headers = {"X-Script-Id": script_id} ) script_cred = response.json().get("data") response = await current_app.http_client.get( url = r"https://nexcom.ditscentre.in/internal/data/get", headers = {"X-Script-Id": script_id} ) current_app.script_data = response.json().get("data") # We set the CPU affinity: try: set_cpu_affinity(script_cred["cpuAffinity"]) except Exception as exception: current_app.printer(exception) # Caching connections: current_app.rate_limit_cache = AsyncRedisCache( connection_string = script_cred["redisCache"]["rateLimit"]["connectionString"], debug = enable_debugging, debug_prefix = "RL Cache | " ) current_app.module_cache = AsyncRedisCache( connection_string = script_cred["redisCache"]["funcReturn"]["connectionString"], serializer = JSONSerializer(), debug = enable_debugging, debug_prefix = "User Cache | " ) # MongoDB connections: current_app.logs_mongo = AsyncMongo( connection_string = script_cred["mongoDb"]["logs"]["connectionString"], database_name = script_cred["mongoDb"]["logs"]["dbName"], max_connections = script_cred["mongoDb"]["logs"]["poolSize"], debug = enable_debugging ) await current_app.logs_mongo.connect() current_app.data_mongo = AsyncMongo( connection_string = script_cred["mongoDb"]["data"]["connectionString"], database_name = script_cred["mongoDb"]["data"]["dbName"], max_connections = script_cred["mongoDb"]["data"]["poolSize"], debug = enable_debugging ) await current_app.data_mongo.connect() # Create an instance to handle GMail-related activities: current_app.gmail_client = AsyncGMailClient( service_name = "gmail", oauth_json = script_cred["google"]["oauth"]["tcaoff"], http_client = current_app.http_client, redirect_url = r"https://api.thecaoffice.com/converse/mail/callback/gmail", scopes = SCOPES_GMAIL_MAIL_MANAGEMENT, debug = True, debug_prefix = "GMail (M) | ", debug_only_errors = False ) # Pick the important stuff: current_app.whitelisted_ips = current_app.script_data["whitelistedIps"] # Register helpers: current_app.get_session = session.get_session # Remove unwanted/sensitive variables from RAM: del script_cred gc.collect() # Done here! current_app.printer("Worker ready!") # --------------------------------------------------------------------------------------------------------------------- @app.after_serving @set_api_version(api_version = APP_VERSION) @log_request_to_mongo( attr_name = "logs_mongo", project = constants.PROJECT_NAME, log_type = constants.MODULE_NAME, operation = "apiStop", log_input = True, log_output = True ) async def app_shutdown(**kwargs): """ This is called when "app.shutdown()" is called. :return: None. """ message = "Shutting down..." current_app.printer(message) # --------------------------------------------------------------------------------------------------------------------- @app.route(f"/", methods = ["GET", "POST"]) @app.route(f"/{MODULE_BASE}", methods = ["GET", "POST"]) async def root(): """ To check if the service is running or not. Use this to monitor the service from your "watchman" script. :return: only "ok" """ return "ok" # --------------------------------------------------------------------------------------------------------------------- @app.route(f"/metrics/memory", methods = ["GET", "POST"]) @app.route(f"/{MODULE_BASE}/metrics/memory", methods = ["GET", "POST"]) async def memory_metrics(): """ TO measure the metrics of the app. :return: A JSON of the metrics. """ # Figure pout the parent process. # This is important for multi-worker environments: parent_pid = os.getppid() parent_process = psutil.Process(parent_pid) parent_mem_info = parent_process.memory_info() # Figure out all the children of the parent process: child_pids = [child.pid for child in parent_process.children()] child_processes = [psutil.Process(child_pid) for child_pid in child_pids] child_stats = [] for pid, process in zip(child_pids, child_processes): mem_info = process.memory_info() child_stats.append({ "pid": pid, "mem": round(mem_info.rss / (1024 ** 2), 2) }) # Construct the response: metrics_json = { "parent": { "pid": parent_pid, "mem": round(parent_mem_info.rss / (1024 ** 2), 2) }, "children": child_stats, "unit": { "mem": "MB" }, "ts": date_time.get_current_ist_date_time(as_string = True) } # Done here: return metrics_json # --------------------------------------------------------------------------------------------------------------------- @app.route(f"/{MODULE_BASE}/debug/", methods = ["POST", "GET"]) async def change_debug(action): """ Enable or disable debugging for the entire microservice. WARNING: NOT RECOMMENDED FOR MULTI-WORKER DEPLOYMENTS. :param action: "enable" to allow debugging on the terminal, or "disable". :return: "enabled"/"disabled" if successful, else "ok" """ # Enable or disable debugging only if the password matches: action = action.lower() if action == "enable": current_app.printer.enable() elif action == "disable": current_app.printer.disable() return "ok" # --------------------------------------------------------------------------------------------------------------------- @app.route(f"/{MODULE_BASE}/maintenance/", methods = ["POST", "GET"]) async def change_maintenance(action): """ Enable or disable debugging for the entire microservice. WARNING: NOT RECOMMENDED FOR MULTI-WORKER DEPLOYMENTS. :param action: "enable" to stop taking new requests on the API, or "disable". :return: "enabled"/"disabled" if successful, else "ok" """ # Enable or disable debugging only if the password matches: action = action.lower() if action == "enable": current_app.is_under_maintenance = True os.environ["IS_UNDER_MAINTENANCE"] = "True" elif action == "disable": current_app.is_under_maintenance = False os.environ["IS_UNDER_MAINTENANCE"] = "False" return "ok" # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": # To get args from the terminal: import argparse # To run the ASGI: import uvicorn from multiprocessing import freeze_support # Get the config from the command-line: parser = argparse.ArgumentParser(description = f"Microservice for '{MODULE_BASE}' API.") parser.add_argument( "--workers", type = int, help = "The no. of threads to spin up for this instance!", default = 2 ) parser.add_argument( "--host", type = str, help = "The host for the app. e.g.: '0.0.0.0' or '127.0.0.1'.", default = "127.0.0.1" ) parser.add_argument( "--port", type = int, help = "The port no. to bind the app to.", default = 8080 ) parser.add_argument( "--script-id", type = str, help = "The id of this script (will affect the loaded config)." ) parser.add_argument( "--debug", action = "store_true", help = "Whether, or not, you want to see debugging messages in the terminal.", default = False ) args = parser.parse_args() # Note down the config; os.environ["SCRIPT_ID"] = args.script_id os.environ["DEBUG"] = str(args.debug) # Run the gateway: freeze_support() uvicorn.run( app = "main:app", workers = args.workers, host = args.host, port = args.port )