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

(20241009) Module ready for use by whitelisted IPs.

Browse Source
This commit is contained in:
khushal
2024-10-09 10:26:32 +05:30
parent 23cfebc20d
commit b366c0aa43
5 changed files with 432 additions and 430 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README_GIT.md
+1 -1
View File
@@ -1,6 +1,6 @@
# GIT CHEATSHEET
## By Sharvil Sir 🙏 (20240927)
## By Sharvil (20240927)
Perform these steps on creating a new project.
_**ASSUMPTION:** You are already in your project directory in the command line terminal._
api/cred_data/blueprint.py
+84 -109
View File
@@ -6,14 +6,14 @@
DATE:
Wednesday, 28th Aug., 2024
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 have anything held in them, the idea behind giving two services is for the user of
this service to be able to organise his setup variables.
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:
@@ -46,15 +46,18 @@ sys.path.append("..")
from quart import Blueprint, current_app
# 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
validate_input,
handle_cancelled_request
)
# For asynchronous activities:
@@ -70,18 +73,14 @@ import asyncio
# Related to Quart:
cred_and_data_bp = Blueprint("int_cnd", __name__)
get_api_version = "2.0.0"
set_api_version = "2.0.0"
update_api_version = "2.0.0"
delete_api_version = "2.0.0"
# related to the operations of this blueprint:
JSON_TYPE_INFO = {
"cred": {
"collection": "scriptCred"
"collection": "_scriptCred"
},
"data": {
"collection": "scriptData"
"collection": "_scriptData"
}
}
@@ -114,94 +113,94 @@ def init(blueprint_setup_state):
@cred_and_data_bp.route("/<json_type>/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",
log_type = "internalCredData",
project = "internal",
log_type = "credData",
operation = "set",
api_version = set_api_version,
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"])
@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,
log_id: str = 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.
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 from 'async_quart_utils.py'.
:param inbound_data: auto-extracted by the decorators from 'async_quart_utils.py'.
:param inbound_files: auto-extracted by the decorators from 'async_quart_utils.py'.
:param log_id: An identifier for the logs (if logging is enabled).
:return: A standard response structure from the function in 'async_quart_utils.py'.
: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.
"""
try:
# If an invalid choice was made:
if json_type not in JSON_TYPE_INFO.keys():
return ResponseModel(
status_code = StatusCodes.FAILED,
message = f"invalid url segment '{json_type}'",
message = f"invalid path '{json_type}'",
http_code = HttpCodes.BAD_REQUEST
)
# Pre-process the inbound data:
inbound_data = inbound_data or {}
inbound_data["scriptId"] = inbound_headers["X-Script-Id"]
# 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 = inbound_data,
upsert = True
replacement = document,
upsert = True,
raise_exception = True
)
# Return the response:
if success: return ResponseModel(api_version = set_api_version, status_code = StatusCodes.OK)
else: return ResponseModel(api_version = set_api_version, status_code = StatusCodes.FAILED)
# In case the client terminates the connection prematurely:
except asyncio.CancelledError as exception:
current_app.printer(exception)
return ResponseModel(
api_version = set_api_version,
status_code = StatusCodes.CLIENT_CLOSED_REQUEST
)
if success: return ResponseModel(status_code = StatusCodes.OK)
else: return ResponseModel(status_code = StatusCodes.FAILED)
# ---------------------------------------------------------------------------------------------------------------------
@cred_and_data_bp.route("/<json_type>/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",
log_type = "internalCredData",
project = "internal",
log_type = "credData",
operation = "get",
api_version = get_api_version,
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,
log_id: str = None
**kwargs
):
"""
@@ -209,20 +208,18 @@ async def get_data(
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 from 'async_quart.py'.
:param inbound_data: auto-extracted by the decorators from 'async_quart.py'.
:param inbound_files: auto-extracted by the decorators from 'async_quart.py'.
:param log_id: An identifier for the logs (if logging is enabled).
:return: A standard response structure from the function in 'async_quart.py'.
: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.
"""
try:
# If an invalid choice was made:
if json_type not in JSON_TYPE_INFO.keys():
return ResponseModel(
status_code = StatusCodes.FAILED,
message = f"invalid url segment '{json_type}'",
message = f"invalid path '{json_type}'",
http_code = HttpCodes.BAD_REQUEST
)
@@ -230,30 +227,21 @@ async def get_data(
cred_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}
projection = {"_id": False, "scriptId": False},
raise_exception = True
)
# In case no result was found:
if cred_json is None:
return ResponseModel(
api_version = get_api_version,
status_code = StatusCodes.FAILED,
message = "invalid script id"
)
# Successfully retrieved:
return ResponseModel(
api_version = get_api_version,
status_code = StatusCodes.OK,
data = cred_json
)
# In case the client terminates the connection prematurely:
except asyncio.CancelledError as exception:
current_app.printer(exception)
return ResponseModel(
api_version = get_api_version,
status_code = StatusCodes.CLIENT_CLOSED_REQUEST
data = cred_json["content"]
)
@@ -261,86 +249,83 @@ async def get_data(
@cred_and_data_bp.route("/<json_type>/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",
log_type = "internalCredData",
project = "internal",
log_type = "credData",
operation = "update",
api_version = update_api_version,
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,
log_id: str = 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 from 'async_quart_utils.py'.
:param inbound_data: auto-extracted by the decorators from 'async_quart_utils.py'.
:param inbound_files: auto-extracted by the decorators from 'async_quart_utils.py'.
:param log_id: An identifier for the logs (if logging is enabled).
:return: A standard response structure from the function in 'async_quart_utils.py'.
: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.
"""
try:
# If an invalid choice was made:
if json_type not in JSON_TYPE_INFO.keys():
return ResponseModel(
status_code = StatusCodes.FAILED,
message = f"invalid url segment '{json_type}'",
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:
# 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(inbound_data["unset"])
if inbound_data.get("set"): update_json["$set"] = current_app.mongo.dict_to_dot_notation(inbound_data["set"])
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
upsert = False,
raise_exception = True
)
# Return the response:
if success: return ResponseModel(api_version = update_api_version, status_code = StatusCodes.OK)
else: return ResponseModel(api_version = update_api_version, status_code = StatusCodes.FAILED)
# In case the client terminates the connection prematurely:
except asyncio.CancelledError as exception:
current_app.printer(exception)
return ResponseModel(
api_version = update_api_version,
status_code = StatusCodes.CLIENT_CLOSED_REQUEST
)
if success: return ResponseModel(status_code = StatusCodes.OK)
else: return ResponseModel(status_code = StatusCodes.FAILED)
# ---------------------------------------------------------------------------------------------------------------------
@cred_and_data_bp.route("/<json_type>/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",
log_type = "internalCredData",
project = "internal",
log_type = "credData",
operation = "delete",
api_version = delete_api_version,
log_input = True,
log_output = True
)
@@ -352,26 +337,24 @@ async def delete_data(
inbound_headers: dict = None,
inbound_data: dict = None,
inbound_files: dict = None,
log_id: str = 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 from 'async_quart.py'.
:param inbound_data: auto-extracted by the decorators from 'async_quart.py'.
:param inbound_files: auto-extracted by the decorators from 'async_quart.py'.
:param log_id: An identifier for the logs (if logging is enabled).
:return: A standard response structure from the function in 'async_quart.py'.
: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.
"""
try:
# If an invalid choice was made:
if json_type not in JSON_TYPE_INFO.keys():
return ResponseModel(
status_code = StatusCodes.FAILED,
message = f"invalid url segment '{json_type}'",
message = f"invalid path '{json_type}'",
http_code = HttpCodes.BAD_REQUEST
)
@@ -382,16 +365,8 @@ async def delete_data(
)
# Return the response:
if success: return ResponseModel(api_version = delete_api_version, status_code = StatusCodes.OK)
else: return ResponseModel(api_version = delete_api_version, status_code = StatusCodes.FAILED)
# In case the client terminates the connection prematurely:
except asyncio.CancelledError as exception:
current_app.printer(exception)
return ResponseModel(
api_version = delete_api_version,
status_code = StatusCodes.CLIENT_CLOSED_REQUEST
)
if success: return ResponseModel(status_code = StatusCodes.OK)
else: return ResponseModel(status_code = StatusCodes.FAILED)
# *****************************************************************************************************************
api/logs/blueprint.py
+180 -205
View File
@@ -6,14 +6,12 @@
DATE:
Created: Wednesday, 28th Aug., 2024
Updated: Tuesday, 8th Oct. 2024
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.
To be able to fetch logs for rapid issue resolution.
REFERENCES:
@@ -72,17 +70,8 @@ import asyncio
# Related to Quart:
cred_and_data_bp = Blueprint("int_cnd", __name__)
logs_bp = Blueprint("int_logs", __name__)
# related to the operations of this blueprint:
JSON_TYPE_INFO = {
"cred": {
"collection": "_scriptCred"
},
"data": {
"collection": "_scriptData"
}
}
# *****************************************************************************************************************
# ***** ****
@@ -101,7 +90,7 @@ JSON_TYPE_INFO = {
# *****************************************************************************************************************
@cred_and_data_bp.record_once
@logs_bp.record_once
def init(blueprint_setup_state):
# This gets called when the blueprint is registered.
@@ -112,159 +101,93 @@ def init(blueprint_setup_state):
# ---------------------------------------------------------------------------------------------------------------------
@cred_and_data_bp.route("/<json_type>/set", methods = ["POST", "GET"])
@logs_bp.route("/get/id/<log_id>", 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,
async def get_log(
log_id,
**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.
To get the log from its log id.
:param log_id: An identifier (string) for the log to fetch.
"""
# If an invalid choice was made:
if json_type not in JSON_TYPE_INFO.keys():
return ResponseModel(
fetched_log = await current_app.mongo.find_one(
collection = "logs",
filter = {"logId": log_id},
projection = {"_id": False}
)
if not fetched_log: return ResponseModel(
status_code = StatusCodes.FAILED,
message = f"invalid path '{json_type}'",
http_code = HttpCodes.BAD_REQUEST
http_code = HttpCodes.NOT_FOUND,
message = "no such log"
)
# 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("/<json_type>/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:
cred_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 cred_json is None:
return ResponseModel(
status_code = StatusCodes.FAILED,
message = "invalid script id"
)
# Successfully retrieved:
return ResponseModel(
else: return ResponseModel(
status_code = StatusCodes.OK,
data = cred_json["content"]
message = f"log found",
data = {"total": 1, "fetched": 1, "logs": fetched_log}
)
# ---------------------------------------------------------------------------------------------------------------------
@cred_and_data_bp.route("/<json_type>/update", methods = ["POST", "GET"])
@logs_bp.route("/get/exception/id/<log_id>", 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,
async def get_exception_from_log(
log_id,
**kwargs
):
"""
To get the log's exception from its log id.
:param log_id: An identifier (string) for the log to fetch.
"""
fetched_log = await current_app.mongo.find_one(
collection = "logs",
filter = {"logId": log_id},
projection = {
"_id": False,
"logId": True,
"log": True,
"operation": True,
"ts": True,
"exception": True
}
)
if not fetched_log: return ResponseModel(
status_code = StatusCodes.FAILED,
http_code = HttpCodes.NOT_FOUND,
message = "no such log"
)
else: return ResponseModel(
status_code = StatusCodes.OK,
message = f"log found",
data = {"total": 1, "fetched": 1, "logs": fetched_log}
)
# ---------------------------------------------------------------------------------------------------------------------
@logs_bp.route("/get/chain/<log_chain>", methods = ["POST", "GET"])
@set_api_version(api_version = "2.1.0")
@read_input(sanitize_headers = False, sanitize_data = False)
@should_not_be_under_maintenance(attr_name = "is_under_maintenance")
@only_whitelisted_ips(attr_name = "whitelisted_ips")
async def get_log_chain(
log_chain,
inbound_headers: dict = None,
inbound_data: dict = None,
inbound_files: dict = None,
@@ -272,68 +195,53 @@ async def update_cred(
):
"""
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'.
To get the series of logs from its chain identifier.
:param log_chain: An identifier (string) for the log chain to fetch.
: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(
total_count = await current_app.mongo.count(
collection = "logs",
filter = {"logChain": log_chain}
)
fetched_logs = await current_app.mongo.find_many(
collection = "logs",
filter = {"logChain": log_chain},
projection = {"_id": False},
sort = inbound_data.get("sort", {"_id": -1}),
limit = inbound_data.get("limit", 25),
skip = inbound_data.get("skip", 0)
)
fetched_count = len(fetched_logs)
if not fetched_logs: return ResponseModel(
status_code = StatusCodes.FAILED,
message = f"invalid path '{json_type}'",
http_code = HttpCodes.BAD_REQUEST
http_code = HttpCodes.NOT_FOUND,
message = "no such log chain"
)
# 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
else: return ResponseModel(
status_code = StatusCodes.OK,
message = f"{fetched_count} logs fetched",
data = {"total": total_count, "fetched": fetched_count, "logs": fetched_logs}
)
# Return the response:
if success: return ResponseModel(status_code = StatusCodes.OK)
else: return ResponseModel(status_code = StatusCodes.FAILED)
# ---------------------------------------------------------------------------------------------------------------------
@cred_and_data_bp.route("/<json_type>/delete", methods = ["POST", "GET"])
@logs_bp.route("/get/exception/chain/<log_chain>", 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
)
@read_input(sanitize_headers = False, sanitize_data = 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"])
async def delete_data(
json_type: str = None,
async def get_exceptions_from_log_chain(
log_chain,
inbound_headers: dict = None,
inbound_data: dict = None,
inbound_files: dict = None,
@@ -341,32 +249,99 @@ async def delete_data(
):
"""
To delete a document for a particular script id.
:param json_type: The choice from one of the fields of 'JSON_TYPE_INFO'.
To get the series of log exceptions from its chain identifier.
:param log_chain: An identifier (string) for the log chain to fetch.
: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(
total_count = await current_app.mongo.count(
collection = "logs",
filter = {"logChain": log_chain}
)
fetched_logs = await current_app.mongo.find_many(
collection = "logs",
filter = {"logChain": log_chain},
projection = {
"_id": False,
"log": True,
"operation": True,
"ts": True,
"exception": True
},
sort = inbound_data.get("sort", {"_id": -1}),
limit = inbound_data.get("limit", 25),
skip = inbound_data.get("skip", 0)
)
fetched_count = len(fetched_logs)
if not fetched_logs: return ResponseModel(
status_code = StatusCodes.FAILED,
message = f"invalid path '{json_type}'",
http_code = HttpCodes.BAD_REQUEST
http_code = HttpCodes.NOT_FOUND,
message = "no such log chain"
)
# 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"]},
else: return ResponseModel(
status_code = StatusCodes.OK,
message = f"{fetched_count} logs fetched",
data = {"total": total_count, "fetched": fetched_count, "logs": fetched_logs}
)
# Return the response:
if success: return ResponseModel(status_code = StatusCodes.OK)
else: return ResponseModel(status_code = StatusCodes.FAILED)
# ---------------------------------------------------------------------------------------------------------------------
@logs_bp.route("/get/filter", methods = ["POST", "GET"])
@set_api_version(api_version = "2.1.0")
@read_input(sanitize_headers = False, sanitize_data = False)
@should_not_be_under_maintenance(attr_name = "is_under_maintenance")
@only_whitelisted_ips(attr_name = "whitelisted_ips")
async def get_logs_by_filter(
inbound_headers: dict = None,
inbound_data: dict = None,
inbound_files: dict = None,
**kwargs
):
"""
To fetch logs by custom filters:
: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.
"""
total_count = await current_app.mongo.count(
collection = "logs",
filter = inbound_data["filter"]
)
fetched_logs = await current_app.mongo.find_many(
collection = "logs",
filter = current_app.mongo.dict_to_dot_notation(inbound_data["filter"]),
projection = inbound_data.get("projection", {"_id": False}),
sort = inbound_data.get("sort", {"_id": -1}),
limit = inbound_data.get("limit", 25),
skip = inbound_data.get("skip", 0)
)
fetched_count = len(fetched_logs)
if not fetched_logs: return ResponseModel(
status_code = StatusCodes.FAILED,
http_code = HttpCodes.NOT_FOUND,
message = "no matching logs"
)
else: return ResponseModel(
status_code = StatusCodes.OK,
message = f"{len(fetched_logs)} / {total_count} logs fetched",
data = {"total": total_count, "fetched": fetched_count, "logs": fetched_logs}
)
# *****************************************************************************************************************
api/main.py
+88 -48
View File
@@ -6,7 +6,7 @@
DATE:
Wednesday, 28th Aug., 2024
Tuesday, 8th Oct., 2024
OBJECTIVE:
@@ -39,37 +39,37 @@ 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
# To make REST-API calls:
import httpx
# 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.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
validate_input,
handle_cancelled_request
)
# 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
# All the blueprints:
from api.cred_data.blueprint import cred_and_data_bp
from api.logs.blueprint import logs_bp
# *****************************************************************************************************************
@@ -80,8 +80,8 @@ from langchain_openai import ChatOpenAI
# Quart related:
MODULE_BASE = "ai"
APP_VERSION = "1.0.0"
MODULE_BASE = "internal"
APP_VERSION = "2.0.0"
# *****************************************************************************************************************
@@ -94,7 +94,8 @@ APP_VERSION = "1.0.0"
# The Quart app:
app = Quart(__name__)
app = cors(app)
app.register_blueprint(llm_chat_bp, url_prefix = f"/{MODULE_BASE}/llm")
app.register_blueprint(cred_and_data_bp, url_prefix = f"/{MODULE_BASE}")
app.register_blueprint(logs_bp, url_prefix = f"/{MODULE_BASE}/logs")
# *****************************************************************************************************************
@@ -105,15 +106,15 @@ app.register_blueprint(llm_chat_bp, url_prefix = f"/{MODULE_BASE}/llm")
@app.before_serving
@set_api_version(api_version = APP_VERSION)
@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():
async def app_startup(**kwargs):
"""
To initialize the variables that you would like to use in this module.
@@ -130,54 +131,41 @@ async def app_startup():
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
connection_string = constants.MONGO_DATA_CONNECTION_STRING,
database_name = constants.MONGO_DATA_DATABASE_NAME,
max_connections = 10,
debug = True
)
# Remove unwanted/sensitive variables from RAM:
del script_cred
gc.collect()
# Get the script data:
script_id = os.environ["SCRIPT_ID"]
current_app.script_data = (await current_app.mongo.find_one(
collection = "_scriptData",
filter = {"scriptId": script_id}
))["content"]
# Pick the important stuff:
current_app.whitelisted_ips = current_app.script_data["whitelistedIps"]
# Done here!
print("Worker ready!")
# ---------------------------------------------------------------------------------------------------------------------
@app.after_serving
@set_api_version(api_version = APP_VERSION)
@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():
async def app_shutdown(**kwargs):
"""
This is called when "app.shutdown()" is called.
@@ -207,6 +195,52 @@ async def root():
# ---------------------------------------------------------------------------------------------------------------------
@app.route(f"/metrics/memory", methods = ["GET", "POST"])
@app.route(f"/{MODULE_BASE}/metrics/memory", methods = ["GET", "POST"])
async def 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/<action>", methods = ["POST", "GET"])
async def change_debug(action):
@@ -274,6 +308,12 @@ if __name__ == "__main__":
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,
@@ -297,5 +337,5 @@ if __name__ == "__main__":
app = "main:app",
workers = args.workers,
host = args.host,
port = 8080
port = args.port
)
cred/README_CERTS.md
+13 -1
View File
@@ -1,6 +1,18 @@
# INSTRUCTIONS TO RENEW CERTIFICATES
**Date:** 2024-09-18
### **Date:** 2024-09-18
---
This project is the core for all projects to be setup smoothly. The idea is to have this one project where we explicitly
point to the credentials to be used for connecting to the database, and then we provide an endpoint from which all other
projects can request for their configuration. This way we make config changes in one place on the database and restart
ech instance of all the other projects to update them.
For example, if we want to update the certificates that another microservice must use, we just update the location path
of that certificate in the database and restart that microservice. That microservice will ask this one where to read the
credentials from.
Perform the following steps to renew the certificates:
1. Go to [jcdev.ditscentre.in/jcdev](http://jcdev.ditscentre.in/jcdev/)
2. Navigate to the respective folders (e.g., `kafka`, `mongo`, etc.).
3. Download and replace the certificates to ensure they have the same names as those already present in the `cred` folder.