""" AUTHOR: Khushal P Soonderji DATE: Created: Wednesday, 18th Sept., 2024 Updated: Tuesday, 8th Oct., 2024 OBJECTIVE: To be able to fetch setup credentials and data for any project. This could include things like default values, URLs to assets, etc. While 'data' and 'cred' can be used interchangeably, the idea behind giving two services is for the user of this service to be able to organise his setup variables. REFERENCES: N/A DOWNLOADS: N/A NOTES: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For using Quart: from quart import Blueprint, current_app, make_response # My utils: from utils_v2.string import json from utils_v2.api.codes import StatusCodes, HttpCodes from utils_v2.api.response import ResponseModel 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 ) # For asynchronous activities: import asyncio # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Related to Quart: cred_and_data_bp = Blueprint("int_cnd", __name__) # related to the operations of this blueprint: JSON_TYPE_INFO = { "cred": { "collection": "_scriptCred" }, "data": { "collection": "_scriptData" } } # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** @cred_and_data_bp.record_once def init(blueprint_setup_state): # This gets called when the blueprint is registered. # Consider this to be a one-time setup for the whole blueprint: pass # --------------------------------------------------------------------------------------------------------------------- @cred_and_data_bp.route("//set", methods = ["POST", "GET"]) @set_api_version(api_version = "2.1.0") @read_input(sanitize_headers = True, sanitize_data = True) @log_request_to_mongo( attr_name = "mongo", project = "internal", log_type = "credData", operation = "set", log_input = False, log_output = True ) @should_not_be_under_maintenance(attr_name = "is_under_maintenance") @only_whitelisted_ips(attr_name = "whitelisted_ips") @validate_input(mandatory_header_keys = ["X-Script-Id", "X-Script-Desc"]) @handle_cancelled_request() async def set_data( json_type: str = None, inbound_headers: dict = None, inbound_data: dict = None, inbound_files: dict = None, **kwargs ): """ To set the credentials for a particular script. If the document exists, it will be overwritten. If the document doesn't exist, it will be created. Ideally use this for only the first time setup. :param json_type: The choice from one of the fields of 'JSON_TYPE_INFO'. :param inbound_headers: auto-extracted by the decorators. :param inbound_data: auto-extracted by the decorators. :param inbound_files: auto-extracted by the decorators. :param kwargs: Any number of extra inputs supplied by the decorators. :return: A standard response structure. """ # If an invalid choice was made: if json_type not in JSON_TYPE_INFO.keys(): return ResponseModel( status_code = StatusCodes.FAILED, message = f"invalid path '{json_type}'", http_code = HttpCodes.BAD_REQUEST ) # Construct the document: description = inbound_headers.get("X-Script-Desc", "") if len(description) > 200: description = description[:200] document = { "scriptId": inbound_headers["X-Script-Id"], "desc": description, "content": inbound_data } # Make an attempt to set the credentials: success = await current_app.mongo.replace_one( collection = JSON_TYPE_INFO[json_type]["collection"], filter = {"scriptId": inbound_headers["X-Script-Id"]}, replacement = document, upsert = True, raise_exception = True ) # Return the response: if success: return ResponseModel(status_code = StatusCodes.OK) else: return ResponseModel(status_code = StatusCodes.FAILED) # --------------------------------------------------------------------------------------------------------------------- @cred_and_data_bp.route("//get", methods = ["POST", "GET"]) @set_api_version(api_version = "2.1.0") @read_input(sanitize_headers = True, sanitize_data = True) @log_request_to_mongo( attr_name = "mongo", project = "internal", log_type = "credData", operation = "get", log_input = True, log_output = False ) @should_not_be_under_maintenance(attr_name = "is_under_maintenance") @only_whitelisted_ips(attr_name = "whitelisted_ips") @validate_input(mandatory_header_keys = ["X-Script-Id"]) @handle_cancelled_request() async def get_data( json_type: str = None, inbound_headers: dict = None, inbound_data: dict = None, inbound_files: dict = None, **kwargs ): """ To retrieve the credentials stored for a specific script. The script's id can be anything set by the programmers. The idea is to have only the script's id stored in the script, and everything else is fetched from the database. This means that we get to store and update everything from one central location. :param json_type: The choice from one of the fields of 'JSON_TYPE_INFO'. :param inbound_headers: auto-extracted by the decorators. :param inbound_data: auto-extracted by the decorators. :param inbound_files: auto-extracted by the decorators. :param kwargs: Any number of extra inputs supplied by the decorators. :return: A standard response structure. """ # If an invalid choice was made: if json_type not in JSON_TYPE_INFO.keys(): return ResponseModel( status_code = StatusCodes.FAILED, message = f"invalid path '{json_type}'", http_code = HttpCodes.BAD_REQUEST ) # Make an attempt to retrieve the credentials: data_json = await current_app.mongo.find_one( collection = JSON_TYPE_INFO[json_type]["collection"], filter = {"scriptId": inbound_headers["X-Script-Id"]}, projection = {"_id": False, "scriptId": False}, raise_exception = True ) # In case no result was found: if data_json is None: return ResponseModel( status_code = StatusCodes.FAILED, message = "invalid script id" ) # Successfully retrieved: response = await make_response(json.to_string(data_json, no_space = True)) response.headers['Content-Type'] = "application/json" return response print("DATA:", dict(data_json)) return json.to_string(data_json, no_space = True) return ResponseModel( status_code = StatusCodes.OK, data = data_json["content"], message = data_json["desc"] ) # --------------------------------------------------------------------------------------------------------------------- @cred_and_data_bp.route("//update", methods = ["POST", "GET"]) @set_api_version(api_version = "2.1.0") @read_input(sanitize_headers = True, sanitize_data = True) @log_request_to_mongo( attr_name = "mongo", project = "internal", log_type = "credData", operation = "update", log_input = False, log_output = True ) @should_not_be_under_maintenance(attr_name = "is_under_maintenance") @only_whitelisted_ips(attr_name = "whitelisted_ips") @validate_input(mandatory_header_keys = ["X-Script-Id"]) @handle_cancelled_request() async def update_cred( json_type: str = None, inbound_headers: dict = None, inbound_data: dict = None, inbound_files: dict = None, **kwargs ): """ To update values of certain fields for a credentials document. It only updates existing values, does NOT add a new document if the document doesn't already exist. :param json_type: The choice from one of the fields of 'JSON_TYPE_INFO'. :param inbound_headers: auto-extracted by the decorators. :param inbound_data: auto-extracted by the decorators. :param inbound_files: auto-extracted by the decorators. :param kwargs: Any number of extra inputs supplied by the decorators. :return: A standard response structure. """ # If an invalid choice was made: if json_type not in JSON_TYPE_INFO.keys(): return ResponseModel( status_code = StatusCodes.FAILED, message = f"invalid path '{json_type}'", http_code = HttpCodes.BAD_REQUEST ) # Pre-process the inbound data: inbound_data = inbound_data or {} # Prepare the update JSON. Pre-process the fields to set and unset. # Our actual data/cred are held inside a field called "content", so we must wrap the request in that: update_json = {} if inbound_data.get("unset"): update_json["$unset"] = current_app.mongo.dict_to_dot_notation({"content": inbound_data["unset"]}) if inbound_data.get("set"): update_json["$set"] = current_app.mongo.dict_to_dot_notation({"content": inbound_data["set"]}) # Make an attempt to set the credentials: success = await current_app.mongo.update_one( collection = JSON_TYPE_INFO[json_type]["collection"], filter = {"scriptId": inbound_headers["X-Script-Id"]}, update = update_json, upsert = False, raise_exception = True ) # Return the response: if success: return ResponseModel(status_code = StatusCodes.OK) else: return ResponseModel(status_code = StatusCodes.FAILED) # --------------------------------------------------------------------------------------------------------------------- @cred_and_data_bp.route("//delete", methods = ["POST", "GET"]) @set_api_version(api_version = "2.1.0") @read_input(sanitize_headers = True, sanitize_data = True) @log_request_to_mongo( attr_name = "mongo", project = "internal", log_type = "credData", operation = "delete", log_input = True, log_output = True ) @should_not_be_under_maintenance(attr_name = "is_under_maintenance") @only_whitelisted_ips(attr_name = "whitelisted_ips") @validate_input(mandatory_header_keys = ["X-Script-Id"]) async def delete_data( json_type: str = None, inbound_headers: dict = None, inbound_data: dict = None, inbound_files: dict = None, **kwargs ): """ To delete a document for a particular script id. :param json_type: The choice from one of the fields of 'JSON_TYPE_INFO'. :param inbound_headers: auto-extracted by the decorators. :param inbound_data: auto-extracted by the decorators. :param inbound_files: auto-extracted by the decorators. :param kwargs: Any number of extra inputs supplied by the decorators. :return: A standard response structure. """ # If an invalid choice was made: if json_type not in JSON_TYPE_INFO.keys(): return ResponseModel( status_code = StatusCodes.FAILED, message = f"invalid path '{json_type}'", http_code = HttpCodes.BAD_REQUEST ) # Make an attempt to set the credentials: success = await current_app.mongo.delete_one( collection = JSON_TYPE_INFO[json_type]["collection"], filter = {"scriptId": inbound_headers["X-Script-Id"]}, ) # Return the response: if success: return ResponseModel(status_code = StatusCodes.OK) else: return ResponseModel(status_code = StatusCodes.FAILED) # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass