""" AUTHOR: Khushal P Soonderji DATE: Wednesday, 28th Aug., 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 # For using Quart: from quart import Quart, request, current_app from quart_cors import cors # To make REST-API calls: import httpx # My utils: from utils_v2.string import json from utils_v2.api import async_quart from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.api.async_quart import ( read_input, log_request_to_mongo, should_not_be_under_maintenance, only_whitelisted_ips, limit_rate, validate_input ) # For debugging: from icecream import IceCreamDebugger # Other blueprints: from api.llm.chat_completion import llm_chat_bp # LangChain-related: from langchain_core.prompts import ChatPromptTemplate from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langchain_openai import ChatOpenAI # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Quart related: MODULE_BASE = "ai" APP_VERSION = "1.0.0" # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # The Quart app: app = Quart(__name__) app = cors(app) app.register_blueprint(llm_chat_bp, url_prefix = f"/{MODULE_BASE}/llm") # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** @app.before_serving @log_request_to_mongo( attr_name = "mongo", log_type = MODULE_BASE, operation = "apiStart", api_version = APP_VERSION, log_input = True, log_output = True ) async def app_startup(): """ 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: current_app.printer = IceCreamDebugger(prefix = f"{MODULE_BASE} (Q) | ", includeContext = True) if os.environ["DEBUG"] == "True": current_app.printer.disable() # To make API calls: max_connections = 5 limits = httpx.Limits( max_keepalive_connections = max_connections, max_connections = max_connections, keepalive_expiry = 3600 ) current_app.http_client = httpx.AsyncClient(limits = limits) # Get the credentials and data for this script: script_id = os.environ.get("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") # To connect to Mongo: current_app.mongo = AsyncMongo( connection_string = script_cred["mongoDb"]["dataDb"]["connectionString"], database_name = script_cred["mongoDb"]["dataDb"]["dbName"], max_connections = script_cred["mongoDb"]["dataDb"]["poolSize"], debug = True if os.environ["DEBUG"] == "True" else False ) # Remove unwanted/sensitive variables from RAM: del script_cred gc.collect() # --------------------------------------------------------------------------------------------------------------------- @app.after_serving @log_request_to_mongo( attr_name = "mongo", log_type = MODULE_BASE, operation = "apiStop", api_version = APP_VERSION, log_input = True, log_output = True ) async def app_shutdown(): """ 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"/{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 elif action == "disable": current_app.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( "--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 = 8080 )