Resetting utils subtree.
This commit is contained in:
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -1,958 +0,0 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Friday, 2nd Aug., 2024
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide an easy way to perform repetitive tasks in quart.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
NOTES:
|
||||
|
||||
01. PLEASE USE "ResponseModel" AS THE RETURNED VALUE OF THE API ENDPOINT IF YOU ARE USING ANY OF THESE
|
||||
DECORATORS.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To work with Quart:
|
||||
from quart import request, current_app, g
|
||||
|
||||
# To make decorators:
|
||||
from functools import wraps
|
||||
|
||||
# My utils:
|
||||
from utils_v2.string import json
|
||||
from utils_v2.date_time import date_time
|
||||
from utils_v2.security import sanitizers
|
||||
from utils_v2.api.codes import StatusCodes, HttpCodes
|
||||
from utils_v2.api.log import APILogModel
|
||||
from utils_v2.api.response import ResponseModel
|
||||
|
||||
# To work with date and time:
|
||||
import time
|
||||
import datetime
|
||||
|
||||
# System-level activities:
|
||||
import io
|
||||
import os
|
||||
|
||||
# For Pydantic data-behaviour_models:
|
||||
import pydantic
|
||||
|
||||
# For hashing and shortening the hash:
|
||||
import hashlib
|
||||
import base64
|
||||
|
||||
# To make things human-readable:
|
||||
import humanize
|
||||
|
||||
# For debugging:
|
||||
import traceback
|
||||
import random
|
||||
import string
|
||||
|
||||
# For asynchronous activities:
|
||||
import asyncio
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# Chars to choose from for random strings:
|
||||
ALPHANUMERIC_CHARS = string.ascii_letters + string.digits
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** EXCEPTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class AuthDetailsIncompleteException(Exception):
|
||||
def __str__(self):
|
||||
return "incomplete auth details"
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
async def data_from_request(inbound_request):
|
||||
|
||||
"""
|
||||
Adaptively extract the params from incoming request in whichever way it was provided.
|
||||
:param inbound_request: The request that came in.
|
||||
:return: A dict (could be empty) of the data/params that came in with the request.
|
||||
"""
|
||||
|
||||
inbound_data = {}
|
||||
|
||||
# Extract data from the params in the URL:
|
||||
from_args = inbound_request.args.to_dict()
|
||||
if isinstance(from_args, dict):
|
||||
for k, v in from_args.items():
|
||||
inbound_data[k] = v
|
||||
|
||||
# Extract data from the raw JSON data:
|
||||
from_json = await request.get_json()
|
||||
if isinstance(from_json, dict):
|
||||
for k, v in from_json.items():
|
||||
inbound_data[k] = v
|
||||
|
||||
# Extract inputs from the form body:
|
||||
from_form = await inbound_request.form
|
||||
from_form = from_form.to_dict()
|
||||
if isinstance(from_form, dict):
|
||||
for k, v in from_form.items():
|
||||
inbound_data[k] = v
|
||||
|
||||
return inbound_data
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def file_from_request(inbound_request, file_key):
|
||||
|
||||
"""
|
||||
Extracts ONE file from the incoming request's form-data.
|
||||
WARNING: NOT RECOMMENDED FOR LARGE FILES. STRICTLY USE FOR SMALL FILES THAT WON'T CRASH THE SCRIPT.
|
||||
:param inbound_request: The incoming request.
|
||||
:param file_key: The key of the file that you want to extract.
|
||||
:return: A tuple of the file's name and data.
|
||||
"""
|
||||
|
||||
file_name = None
|
||||
file_data = None
|
||||
files = await inbound_request.files
|
||||
|
||||
if file_key in files:
|
||||
file_name = files[file_key].filename
|
||||
file_data = io.BytesIO(files[file_key].read())
|
||||
|
||||
return file_name, file_data
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def files_from_request(inbound_request: request):
|
||||
|
||||
"""
|
||||
Extracts ALL files from the incoming request's form-data.
|
||||
WARNING: NOT RECOMMENDED FOR LARGE FILES. STRICTLY USE FOR SMALL FILES THAT WON'T CRASH THE SCRIPT.
|
||||
:param inbound_request: The incoming request.
|
||||
:return: A dict describing the file's name, data, and size.
|
||||
"""
|
||||
|
||||
files = await inbound_request.files
|
||||
|
||||
inbound_files = {}
|
||||
for file_key in files:
|
||||
file_data = io.BytesIO(files[file_key].read())
|
||||
file_size = file_data.seek(0, 2)
|
||||
file_data.seek(0)
|
||||
inbound_files[file_key] = {
|
||||
"name": files[file_key].filename,
|
||||
"data": file_data,
|
||||
"size": file_size,
|
||||
"type": files[file_key].content_type
|
||||
}
|
||||
|
||||
return inbound_files
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def headers_from_request(
|
||||
inbound_request,
|
||||
mandatory_keys: list = None
|
||||
):
|
||||
|
||||
"""
|
||||
Extract custom headers and some extra info. from the incoming request.
|
||||
Raises an exception if any mandatory key is missing.
|
||||
IMPORTANT: CUSTOMIZE THIS FOR THE NEEDS OF YOUR PROJECT.
|
||||
:param inbound_request: The request that came in.
|
||||
:param mandatory_keys: The keys that you need to have in the auth.
|
||||
:return: The extracted auth details.
|
||||
"""
|
||||
|
||||
# Start by extracting whatever complies with the format of "X-{Header-Name}":
|
||||
head_json = {k: v for k, v in request.headers.items() if k.startswith("X-")}
|
||||
|
||||
# Now note down things that you want to keep from incoming requests:
|
||||
head_json["Remote-IP"] = inbound_request.remote_addr
|
||||
head_json["Host"] = inbound_request.headers.get("Host")
|
||||
head_json["Origin"] = inbound_request.headers.get("Origin")
|
||||
head_json["User-Agent"] = inbound_request.headers.get("User-Agent")
|
||||
|
||||
# Raise an exception if any of the mandatory auth details were missing:
|
||||
if mandatory_keys is not None:
|
||||
available_keys = head_json.keys()
|
||||
for mandatory_key in mandatory_keys:
|
||||
if mandatory_key not in available_keys: raise AuthDetailsIncompleteException
|
||||
|
||||
# Done here:
|
||||
return head_json
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def cause_exception():
|
||||
|
||||
"""
|
||||
Call this from any function when you want to raise an exception.
|
||||
Example use case would be when receiving data from an API call and that field is not supposed to be null.
|
||||
:return: None.
|
||||
"""
|
||||
|
||||
return 100/0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def describe_exception(exc):
|
||||
|
||||
"""
|
||||
Describes the exception in detail. It extracts the type of exception, a brief message, and even the entire
|
||||
traceback. Useful for debugging in details without the terminal. You could either log the resultant dict or send it
|
||||
to the dev team over some service like WhatsApp/Telegram.
|
||||
:param exc: The exception that occurred.
|
||||
:return: The dict that explains the exception.
|
||||
"""
|
||||
|
||||
exc_desc = {
|
||||
"type": type(exc).__name__,
|
||||
"msg": str(exc),
|
||||
"tb": [str(exc_tb) for exc_tb in traceback.format_exception(exc, value = exc, tb = exc.__traceback__)]
|
||||
}
|
||||
|
||||
return exc_desc
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def messages_from_pydantic_exception(exception, as_str = True, sep = ", "):
|
||||
|
||||
"""
|
||||
Creates a list of readable error messages from Pydantic's validation failure.
|
||||
:param exception: Pydantic's ValidationError
|
||||
:param as_str: Set to True to receive all messages as one string, False to receive an array of strings.
|
||||
:param sep: The separator to use when joining multiple messages as one string.
|
||||
:return: A list of messages of all the things that went wrong.
|
||||
"""
|
||||
|
||||
# Make a variable to hold all individual messages:
|
||||
messages = []
|
||||
|
||||
# Interpret all the problems:
|
||||
for error in exception.errors():
|
||||
loc = " --> ".join([str(item) for item in error["loc"]])
|
||||
if error["type"] == "missing": messages.append(f"missing input: {loc}")
|
||||
elif error["type"] == "model_type": messages.append(f"invalid input: {loc}")
|
||||
elif error["type"] == "bool_parsing": messages.append(f"invalid bool: {loc}")
|
||||
elif error["type"] == "string_type": messages.append(f"invalid string: {loc}")
|
||||
elif error["type"] == "float_parsing": messages.append(f"invalid float: {loc}")
|
||||
elif error["type"] == "int_parsing": messages.append(f"invalid integer: {loc}")
|
||||
elif error["type"] == "extra_forbidden": messages.append(f"extra input: {loc}")
|
||||
elif error["type"] == "value_error": messages.append(f"validation failed: {loc}")
|
||||
else: messages.append(f"invalid datatype: {loc}")
|
||||
|
||||
# Return a response as per the preference of the user:
|
||||
if as_str: return sep.join(messages)
|
||||
else: return messages
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def set_api_version(api_version):
|
||||
|
||||
"""
|
||||
Use this decorator to automatically note down the API version no. and propagate it throughout the downstream
|
||||
decorators. Use this as the entry point if possible.
|
||||
:param api_version: The version code to assign to the API.
|
||||
:return: The decorator factory.
|
||||
"""
|
||||
|
||||
def decorator(func):
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
|
||||
# Let the next in-line decorator know that it has been wrapped:
|
||||
kwargs["decorator_count"] = kwargs.get("decorator_count", 0) + 1
|
||||
|
||||
# set the version information in the variable.
|
||||
# This makes it available to the downstream decorators too!
|
||||
kwargs["api_version"] = api_version
|
||||
|
||||
# we are ready to call the function that we are wrapping:
|
||||
response = await func(*args, **kwargs)
|
||||
kwargs["decorator_count"] -= 1
|
||||
if isinstance(response, ResponseModel): response.api_version = api_version
|
||||
|
||||
# Done here:
|
||||
if (
|
||||
kwargs["decorator_count"] == 0 and
|
||||
isinstance(response, ResponseModel)
|
||||
): response = response.for_quart()
|
||||
return response
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def read_input(
|
||||
sanitize_headers = True,
|
||||
sanitize_data = True
|
||||
):
|
||||
|
||||
"""
|
||||
Use this decorator to read the inputs from the incoming request and sanitize them. Sanitization makes the inputs
|
||||
safe against certain threats like injections attacks. If you expect to take in inputs that you want to use to run
|
||||
database commands, you could disable them manually.
|
||||
PLEASE USE "ResponseModel" AS THE RETURNED VALUE OF THE API ENDPOINT IF YOU ARE USING THIS DECORATOR.
|
||||
:param sanitize_headers: Whether, or not, you would like to sanitize the params coming in through the headers.
|
||||
:param sanitize_data: Whether, or not, you would like to sanitize the params coming in through the body or query.
|
||||
:return: The decorator factory.
|
||||
"""
|
||||
|
||||
def decorator(func):
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
|
||||
# Let the next in-line decorator know that it has been wrapped:
|
||||
kwargs["decorator_count"] = kwargs.get("decorator_count", 0) + 1
|
||||
|
||||
# Get the headers:
|
||||
kwargs["inbound_headers"] = await headers_from_request(request)
|
||||
if sanitize_headers: kwargs["inbound_headers"] = sanitizers.for_mongo(kwargs["inbound_headers"])
|
||||
|
||||
# Get the data:
|
||||
kwargs["inbound_data"] = await data_from_request(request)
|
||||
if sanitize_data: kwargs["inbound_data"] = sanitizers.for_mongo(kwargs["inbound_data"])
|
||||
|
||||
# Get small files from the request:
|
||||
kwargs["inbound_files"] = await files_from_request(request)
|
||||
|
||||
# We also make a provision for capturing an identifier
|
||||
# for the logs that we make through a sister decorator:
|
||||
kwargs["log_id"] = "".join(random.choice(ALPHANUMERIC_CHARS) for _ in range(8))
|
||||
|
||||
# Now that we have unpacked the incoming data,
|
||||
# we are ready to run the function that we are wrapping:
|
||||
response = await func(*args, **kwargs)
|
||||
kwargs["decorator_count"] -= 1
|
||||
|
||||
# Done here:
|
||||
if (
|
||||
kwargs["decorator_count"] == 0 and
|
||||
isinstance(response, ResponseModel)
|
||||
): response = response.for_quart()
|
||||
return response
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def validate_input(
|
||||
mandatory_header_keys = None,
|
||||
mandatory_data_keys = None,
|
||||
mandatory_file_keys = None,
|
||||
header_validator = None,
|
||||
data_validator = None
|
||||
):
|
||||
|
||||
"""
|
||||
USE THIS ONLY AFTER YOU HAVE USED 'read_input'. This decorator will help you run validation on the inputs that were
|
||||
extracted from the request. The mandatory keys will be checked first and the validations will be run after that. If
|
||||
your validator already checks for keys, you may skip mentioning mandatory keys. DO NOTE THAT YOUR VALIDATOR
|
||||
FUNCTIONS MUST RAISE AN EXCEPTION FOR THIS DECORATOR TO WORK.
|
||||
:param mandatory_header_keys: The keys in 'inbound_headers' that are absolutely necessary.
|
||||
:param mandatory_data_keys: The keys in 'inbound_data' that are absolutely necessary.
|
||||
:param mandatory_file_keys: The keys in 'inbound_files' that are absolutely necessary.
|
||||
:param header_validator: The function to use to validate the 'inbound_headers'.
|
||||
:param data_validator: The function to use to validate 'inbound_data'.
|
||||
:return: The decorator factory.
|
||||
"""
|
||||
|
||||
def decorator(func):
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
|
||||
# Let the next in-line decorator know that it has been wrapped:
|
||||
kwargs["decorator_count"] = kwargs.get("decorator_count", 0) + 1
|
||||
|
||||
# We first validate the mandatory header keys.
|
||||
# Having a null value in this case is NOT allowed:
|
||||
if mandatory_header_keys is not None:
|
||||
for mandatory_key in mandatory_header_keys:
|
||||
if kwargs["inbound_headers"].get(mandatory_key) is None:
|
||||
return ResponseModel(
|
||||
status_code = StatusCodes.HEADERS_INCOMPLETE,
|
||||
message = f"missing: '{mandatory_key}'"
|
||||
)
|
||||
|
||||
# Return with failure if any of the mandatory JSON details are missing.
|
||||
# Having a null value is allowed, but is should be sent by the user on intention.
|
||||
if mandatory_data_keys is not None:
|
||||
for mandatory_key in mandatory_data_keys:
|
||||
try: kwargs["inbound_data"][mandatory_key]
|
||||
except: return ResponseModel(
|
||||
status_code = StatusCodes.DATA_INCOMPLETE,
|
||||
message = f"missing: '{mandatory_key}'"
|
||||
)
|
||||
|
||||
# Return with failure if any of the mandatory file-keys details are missing:
|
||||
if mandatory_file_keys is not None:
|
||||
provided_file_keys = kwargs["inbound_files"].keys()
|
||||
for mandatory_key in mandatory_file_keys:
|
||||
if mandatory_key not in provided_file_keys:
|
||||
return ResponseModel(
|
||||
status_code = StatusCodes.FILE_MISSING,
|
||||
message = f"missing: '{mandatory_key}'"
|
||||
)
|
||||
|
||||
# Next we validate the headers:
|
||||
if header_validator is not None:
|
||||
|
||||
# Try validate the data:
|
||||
try: kwargs["inbound_headers"] = header_validator(kwargs["inbound_headers"])
|
||||
|
||||
# In case some needed field is missing:
|
||||
except KeyError as exception:
|
||||
return ResponseModel(
|
||||
status_code = StatusCodes.DATA_VALIDATION_FAILURE,
|
||||
message = "missing: " + str(exception),
|
||||
http_code = HttpCodes.BAD_REQUEST,
|
||||
)
|
||||
|
||||
# In case some pydantic data model fails validation:
|
||||
except pydantic.ValidationError as exception:
|
||||
return ResponseModel(
|
||||
status_code = StatusCodes.DATA_VALIDATION_FAILURE,
|
||||
message = messages_from_pydantic_exception(exception),
|
||||
http_code = HttpCodes.BAD_REQUEST
|
||||
)
|
||||
|
||||
# In case some other exception was raised:
|
||||
except Exception as exception:
|
||||
return ResponseModel(
|
||||
status_code = StatusCodes.DATA_VALIDATION_FAILURE,
|
||||
message = str(exception),
|
||||
http_code = HttpCodes.BAD_REQUEST
|
||||
)
|
||||
|
||||
# Finally, we validate the incoming data:
|
||||
if data_validator is not None:
|
||||
|
||||
# Try validate the data:
|
||||
try: kwargs["inbound_data"] = data_validator(kwargs["inbound_data"])
|
||||
|
||||
# In case some needed field is missing:
|
||||
except KeyError as exception:
|
||||
return ResponseModel(
|
||||
status_code = StatusCodes.DATA_VALIDATION_FAILURE,
|
||||
message = "missing: " + str(exception),
|
||||
http_code = HttpCodes.BAD_REQUEST
|
||||
)
|
||||
|
||||
# In case some pydantic data model fails validation:
|
||||
except pydantic.ValidationError as exception:
|
||||
return ResponseModel(
|
||||
status_code = StatusCodes.DATA_VALIDATION_FAILURE,
|
||||
message = messages_from_pydantic_exception(exception),
|
||||
http_code = HttpCodes.BAD_REQUEST
|
||||
)
|
||||
|
||||
# In case some other exception was raised:
|
||||
except Exception as exception:
|
||||
return ResponseModel(
|
||||
status_code = StatusCodes.DATA_VALIDATION_FAILURE,
|
||||
message = str(exception),
|
||||
http_code = HttpCodes.BAD_REQUEST
|
||||
)
|
||||
|
||||
# Now that we have unpacked the incoming data,
|
||||
# we are ready to run the function that we are wrapping:
|
||||
response = await func(*args, **kwargs)
|
||||
kwargs["decorator_count"] -= 1
|
||||
|
||||
# Done here:
|
||||
if (
|
||||
kwargs["decorator_count"] == 0 and
|
||||
isinstance(response, ResponseModel)
|
||||
): response = response.for_quart()
|
||||
return response
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def log_request_to_mongo(
|
||||
attr_name,
|
||||
collection: str = "logs",
|
||||
api_version: str = None,
|
||||
project: str = None,
|
||||
log_type: str = None,
|
||||
operation: str = None,
|
||||
log_input: bool = True,
|
||||
log_output: bool = True,
|
||||
sensitive_keys: list = None
|
||||
):
|
||||
|
||||
"""
|
||||
USE THIS ONLY AFTER YOU HAVE USED 'read_input'. This decorator will log the whole process of the API call to
|
||||
MongoDB. The variable that holds the instance of 'AsyncMongo' needs to be accessible in the scope of 'current_app'.
|
||||
PLEASE USE "ResponseModel" AS THE RETURNED VALUE OF THE API ENDPOINT IF YOU ARE USING THIS DECORATOR.
|
||||
:param attr_name: The name of the variable that holds the instance of 'AsyncMongo'. It should be accessible in the
|
||||
scope of 'current_app'.
|
||||
:param collection: The name of the collection to write the log into.
|
||||
:param api_version: The version code of the API endpoint that is being logged.
|
||||
:param project: The name of the project that the endpoint was built for.
|
||||
:param log_type: A hint to identify what the log was for.
|
||||
:param operation: A hint to identify what was action was being performed.
|
||||
:param log_input: Whether, or not, you would like to log the input that came in.
|
||||
:param log_output: Whether, or not, you would like to log the output of the API call.
|
||||
:param sensitive_keys: The list of keys to not log.
|
||||
:return: The decorator factory.
|
||||
"""
|
||||
|
||||
def decorator(func):
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
|
||||
# Let the next in-line decorator know that it has been wrapped:
|
||||
kwargs["decorator_count"] = kwargs.get("decorator_count", 0) + 1
|
||||
|
||||
# Set the api version as needed:
|
||||
kwargs["api_version"] = kwargs.get("api_version", api_version)
|
||||
|
||||
# Make variables and extract available info.:
|
||||
exception = None
|
||||
response = None
|
||||
request_ts = date_time.get_current_utc_date_time()
|
||||
start_ts = time.perf_counter()
|
||||
cpu_start_ts = time.process_time()
|
||||
|
||||
# Execute the function that is being wrapped:
|
||||
try:
|
||||
response = await func(*args, **kwargs)
|
||||
kwargs["decorator_count"] -= 1
|
||||
except Exception as exc: exception = exc
|
||||
|
||||
# Ensure that the response is not null:
|
||||
response = response if response is not None else ResponseModel(
|
||||
status_code = StatusCodes.UNKNOWN_ERROR,
|
||||
message = "null response for request"
|
||||
)
|
||||
|
||||
# Add params to the response.
|
||||
# THIS IS ONLY APPLICABLE WHEN THE TYPE OF THE RESPONSE IS 'ResponseModel':
|
||||
if isinstance(response, ResponseModel):
|
||||
response.api_version = kwargs.get("api_version")
|
||||
response.log_id = kwargs.get("log_id")
|
||||
|
||||
# Extract the response to log:
|
||||
response_to_log = "not logged"
|
||||
http_code_to_log = 200
|
||||
if log_output:
|
||||
if isinstance(response, ResponseModel): response_to_log, http_code_to_log = response.for_quart()
|
||||
elif isinstance(response, tuple): response_to_log, http_code_to_log = response
|
||||
else: response_to_log, http_code_to_log = str(response), 200
|
||||
|
||||
# Try to get the information about the request.
|
||||
# There will be no data in any of these if the decorator was used to catch start-up and shut-down events.
|
||||
request_method = None
|
||||
request_url = None
|
||||
request_route = None
|
||||
try:
|
||||
request_method = f"{request.method}"
|
||||
request_url = f"{request.url}"
|
||||
request_route = str(request.url_rule.rule)
|
||||
except: pass
|
||||
|
||||
# Redact the sensitive keys:
|
||||
if sensitive_keys:
|
||||
for k in sensitive_keys:
|
||||
for var in ["inbound_headers", "inbound_data"]:
|
||||
try: kwargs[var][k] = len(str(kwargs[var][k])) * "*"
|
||||
except: pass
|
||||
|
||||
# Construct the log:
|
||||
api_log = APILogModel(
|
||||
project = project,
|
||||
log = log_type,
|
||||
operation = operation,
|
||||
apiVer = kwargs.get("api_version"),
|
||||
logId = kwargs.get("log_id"),
|
||||
logChain = kwargs.get("inbound_headers", {}).get("X-Log-Chain"),
|
||||
method = request_method,
|
||||
url = request_url,
|
||||
route = request_route,
|
||||
ts = request_ts,
|
||||
tat = time.perf_counter() - start_ts,
|
||||
cpuTime = time.process_time() - cpu_start_ts,
|
||||
headers = kwargs.get("inbound_headers"),
|
||||
data = kwargs.get("inbound_data") if log_input else "not logged",
|
||||
files = {
|
||||
k: {
|
||||
"name": v["name"],
|
||||
"size": v["size"]
|
||||
} for k, v in kwargs.get("inbound_files", {}).items()
|
||||
},
|
||||
exception = None if exception is None else describe_exception(exception),
|
||||
response = response_to_log,
|
||||
httpCode = http_code_to_log
|
||||
)
|
||||
|
||||
# Write the log:
|
||||
app_attr = getattr(current_app, attr_name)
|
||||
inserted_id = await app_attr.insert_one(
|
||||
collection = collection,
|
||||
document = api_log.model_dump()
|
||||
)
|
||||
|
||||
# Return the response from the wrapped function.
|
||||
if (
|
||||
kwargs["decorator_count"] == 0 and
|
||||
isinstance(response, ResponseModel)
|
||||
): response = response.for_quart()
|
||||
return response
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def should_not_be_under_maintenance(attr_name):
|
||||
|
||||
"""
|
||||
Use this decorator to reject a request when the app is being marked as "under-maintenance". You will need to create
|
||||
a boolean variable within the scope of the 'current_app' for this to work. An alternate to this is to set the value
|
||||
in an environment variable named 'IS_UNDER_MAINTENANCE' to a string value of either 'True' or 'False' for
|
||||
multi-worker deployments.
|
||||
:param attr_name: The name of the boolean variable that will hold the information about the app being under
|
||||
maintenance. If its value is True at the time of checking, the incoming request will be rejected.
|
||||
:return: The decorator factory.
|
||||
"""
|
||||
|
||||
def decorator(func):
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
|
||||
# Let the next in-line decorator know that it has been wrapped:
|
||||
kwargs["decorator_count"] = kwargs.get("decorator_count", 0) + 1
|
||||
|
||||
# Get the attribute and check if it indicates that the app is under maintenance,
|
||||
# call the wrapped function if not under maintenance:
|
||||
app_attr = getattr(current_app, attr_name)
|
||||
env_attr = True if os.environ.get("IS_UNDER_MAINTENANCE", "False").lower() == "true" else False
|
||||
if app_attr or env_attr:
|
||||
response = ResponseModel(status_code = StatusCodes.DOWN_FOR_MAINTENANCE).for_quart()
|
||||
else:
|
||||
response = await func(*args, **kwargs)
|
||||
kwargs["decorator_count"] -= 1
|
||||
|
||||
# Done here:
|
||||
if (
|
||||
kwargs["decorator_count"] == 0 and
|
||||
isinstance(response, ResponseModel)
|
||||
): response = response.for_quart()
|
||||
return response
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def only_whitelisted_ips(attr_name):
|
||||
|
||||
"""
|
||||
Use this decorator to reject any requests coming from unauthorized IPs. The list of IP addresses to allow must be
|
||||
in a list that is accessible in the context of 'current_app'.
|
||||
:param attr_name: The name of the boolean variable that will hold the information about the app being under
|
||||
maintenance. If its value is True at the time of checking, the incoming request will be rejected.
|
||||
:return: The decorator factory.
|
||||
"""
|
||||
|
||||
def decorator(func):
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
|
||||
# Let the next in-line decorator know that it has been wrapped:
|
||||
kwargs["decorator_count"] = kwargs.get("decorator_count", 0) + 1
|
||||
|
||||
# Get the attribute and check if the request's IP is in the permitted list:
|
||||
app_attr = getattr(current_app, attr_name)
|
||||
if request.remote_addr not in app_attr:
|
||||
return ResponseModel(
|
||||
status_code = StatusCodes.AUTHORIZATION_FAILED,
|
||||
message = "bad ip",
|
||||
http_code = HttpCodes.UNAUTHORIZED
|
||||
).for_quart()
|
||||
|
||||
# Now that we have checked that the IP is permitted,
|
||||
# we are ready to run the function that we are wrapping:
|
||||
response = await func(*args, **kwargs)
|
||||
kwargs["decorator_count"] -= 1
|
||||
|
||||
# Done here:
|
||||
if (
|
||||
kwargs["decorator_count"] == 0 and
|
||||
isinstance(response, ResponseModel)
|
||||
): response = response.for_quart()
|
||||
return response
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def limit_rate(
|
||||
attr_name,
|
||||
rate_limit: int = 5,
|
||||
seconds: float = 1.0,
|
||||
message = None,
|
||||
header_keys: list = None,
|
||||
data_keys: list = None,
|
||||
allow_if_exception = False,
|
||||
count_for_http_codes = None
|
||||
):
|
||||
|
||||
"""
|
||||
Use this decorator to apply rate-limiting to incoming requests.
|
||||
SUGGESTION: WHEN STACKING UP MANY RATE LIMITS, PUT THE SMALLEST TIME PERIOD ON TOP AND LARGEST TIME PERIOD AT THE
|
||||
BOTTOM. THIS ENSURES PROPER FUNCTIONALITY.
|
||||
WARNING: TIMING STARTS WHEN THE FIRST PERMITTED REQUEST GOES THROUGH. THIS MEANS THAT, IF YOU HAVE A PER-DAY LIMIT,
|
||||
AND YOU START MAKING REQUESTS AT 11:00 PM AND EXHAUST YOUR LIMIT AT 11:59 PM, YOUR LIMIT WILL BE REPLENISHED AT
|
||||
11:00 PM OF THE NEXT DAY, NOT AT 12:00 AM.
|
||||
:param attr_name: The name of the variable that holds the instance of 'AsyncRedisCache'. Should be available in the
|
||||
context of 'current_app'.
|
||||
:param rate_limit: The number of requests per unit time.
|
||||
:param seconds: The time period in which the rate limit is to be applied.
|
||||
:param message: The custom message to respond with.
|
||||
:param header_keys: The keys in the header to consider when apply rate limits (like 'sessionToken').
|
||||
:param data_keys: The keys in the header to consider when apply rate limits (like 'sessionToken').
|
||||
:param allow_if_exception: In case Redis is unresponsive, would you prefer allowing the request to pass through or
|
||||
would you prefer the request getting blocked.
|
||||
:param count_for_http_codes: If this is provided, the counter will be incremented only if the response code was one
|
||||
of these values. If not provided, all requests will be counted. This can be used in cases when you want to count
|
||||
only when the request was successfully served.
|
||||
:return: The decorator factory.
|
||||
"""
|
||||
|
||||
# Param-cleaning:
|
||||
rate_limit = max(1, rate_limit)
|
||||
if header_keys is None: header_keys = []
|
||||
if data_keys is None: data_keys = []
|
||||
|
||||
def decorator(func):
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
|
||||
# Let the next in-line decorator know that it has been wrapped:
|
||||
kwargs["decorator_count"] = kwargs.get("decorator_count", 0) + 1
|
||||
|
||||
# Note down the combination of values requested and the limits prescribed:
|
||||
params = {"route": str(request.url_rule), "header": {}, "data": {}, "limit": rate_limit, "seconds": seconds}
|
||||
for key in header_keys: params["header"][key] = kwargs["inbound_headers"].get(key)
|
||||
for key in data_keys: params["data"][key] = kwargs["inbound_data"].get(key)
|
||||
|
||||
# Now make a unique key from this combination:
|
||||
params_json = json.to_string(params, no_space = True)
|
||||
sha256_hash = hashlib.sha256()
|
||||
sha256_hash.update(params_json.encode("utf-8"))
|
||||
hashed_key = sha256_hash.digest()
|
||||
base64_key = base64.b64encode(hashed_key).decode("utf-8")
|
||||
|
||||
# We first get the value of the counter:
|
||||
app_attr = getattr(current_app, attr_name)
|
||||
counter_value = await app_attr.count(base64_key, value = 1, expiry = seconds)
|
||||
|
||||
# If any exception occurred in getting the count,
|
||||
# and exceptions haven't been allowed:
|
||||
if counter_value is None and not allow_if_exception:
|
||||
response = ResponseModel(
|
||||
status_code = StatusCodes.RATE_LIMIT_EXCEEDED,
|
||||
message = "Please contact admin (E)"
|
||||
)
|
||||
|
||||
# If the rate-limit has already been crossed,
|
||||
# or when the counter was not fetched but exceptions are allowed:
|
||||
elif (counter_value or 0) > rate_limit:
|
||||
response = ResponseModel(
|
||||
status_code = StatusCodes.RATE_LIMIT_EXCEEDED,
|
||||
message = message or ", ".join([
|
||||
f"rate limit: {rate_limit} in {humanize.naturaldelta(datetime.timedelta(seconds = seconds))}",
|
||||
f"this is your {humanize.ordinal(counter_value)} request in the given period"
|
||||
])
|
||||
)
|
||||
|
||||
# If the rate-limit hasn't been crossed:
|
||||
else:
|
||||
response = await func(*args, **kwargs)
|
||||
kwargs["decorator_count"] -= 1
|
||||
|
||||
# If we have been told to count only for specific status codes,
|
||||
# and if the HTTP code of this response is not in the list of codes, we reduce the counter by one:
|
||||
if count_for_http_codes:
|
||||
http_code = 200 if not isinstance(response, (list, tuple, set)) else response[1]
|
||||
if http_code not in count_for_http_codes:
|
||||
await app_attr.count(base64_key, value = -1, expiry = seconds)
|
||||
|
||||
# Return the response from the function call:
|
||||
if (
|
||||
kwargs["decorator_count"] == 0 and
|
||||
isinstance(response, ResponseModel)
|
||||
): response = response.for_quart()
|
||||
return response
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def handle_cancelled_request(cleanup_func = None, cleanup_coro = None):
|
||||
|
||||
"""
|
||||
Use this decorator to handle prematurely terminated requests. If your clean-up function needs access to variables,
|
||||
consider using 'g' to hold data in the scope of the request.
|
||||
:param cleanup_func: The function to call when the cancelled request needs graceful handling.
|
||||
:param cleanup_coro: The coroutine to call when the cancelled request needs graceful handling.
|
||||
:return: The decorator factory.
|
||||
"""
|
||||
|
||||
def decorator(func):
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
|
||||
# Let the next in-line decorator know that it has been wrapped:
|
||||
kwargs["decorator_count"] = kwargs.get("decorator_count", 0) + 1
|
||||
|
||||
try:
|
||||
|
||||
# we are ready to run the function that we are wrapping:
|
||||
response = await func(*args, **kwargs)
|
||||
kwargs["decorator_count"] -= 1
|
||||
|
||||
# Done here:
|
||||
if (
|
||||
kwargs["decorator_count"] == 0 and
|
||||
isinstance(response, ResponseModel)
|
||||
): response = response.for_quart()
|
||||
return response
|
||||
|
||||
# In case the client closes the connection pre-maturely:
|
||||
except asyncio.CancelledError as exception:
|
||||
if hasattr(current_app, "printer"): getattr(current_app, "printer")(exception)
|
||||
if cleanup_func is not None: cleanup_func()
|
||||
if cleanup_coro is not None: await cleanup_coro()
|
||||
return ResponseModel(
|
||||
api_version = kwargs.get("api_version"),
|
||||
status_code = StatusCodes.CLIENT_CLOSED_REQUEST
|
||||
).for_quart()
|
||||
|
||||
# We propagate any other kind of exception:
|
||||
except Exception as exception: raise
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
pass
|
||||
@@ -1,164 +0,0 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Thursday, 12th Sept., 2024
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To maintain all status codes in one place.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
from enum import Enum, unique
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
@unique
|
||||
class HttpCodes(Enum):
|
||||
|
||||
"""
|
||||
Commonly used standard HTTP status codes.
|
||||
Can be sent after an API Call is processed.
|
||||
Refer to: https://http.dev/status
|
||||
NOTE: THIS LIST IS NOT EXHAUSTIVE!
|
||||
"""
|
||||
|
||||
# 1XX - Informational:
|
||||
CONTINUE = 100 # .............. https://http.dev/102
|
||||
SWITCHING_PROTOCOLS = 101 # ... https://http.dev/101
|
||||
PROCESSING = 102 # ............ https://http.dev/102
|
||||
EARLY_HINTS = 103 # ........... https://http.dev/103
|
||||
|
||||
# 2XX - Success:
|
||||
SUCCESS = 200 # .................. https://http.dev/200
|
||||
CREATED = 201 # .................. https://http.dev/201
|
||||
ACCEPTED = 202 # ................. https://http.dev/202
|
||||
NON_AUTHORITATIVE_INFO = 203 # ... https://http.dev/203
|
||||
NO_CONTENT = 204 # ............... https://http.dev/204
|
||||
RESET_CONTENT = 205 # ............ https://http.dev/205
|
||||
PARTIAL_CONTENT = 206 # .......... https://http.dev/206
|
||||
MULTI_STATUS = 207 # ............. https://http.dev/207
|
||||
ALREADY_REPORTED = 208 # ......... https://http.dev/208
|
||||
THIS_IS_FINE = 218 # ............. https://http.dev/218
|
||||
IM_USED = 226 # .................. https://http.dev/226
|
||||
|
||||
# 3XX - Redirection:
|
||||
MULTIPLE_CHOICES = 300 # ..... https://http.dev/300
|
||||
MOVED_PERMANENTLY = 301 # .... https://http.dev/301
|
||||
MOVED_TEMPORARILY = 302 # .... https://http.dev/302
|
||||
SEE_OTHER = 303 # ............ https://http.dev/303
|
||||
NOT_MODIFIED = 304 # ......... https://http.dev/304
|
||||
USE_PROXY = 305 # ............ https://http.dev/305
|
||||
SWITCH_PROXY = 306 # ......... https://http.dev/306
|
||||
TEMPORARY_REDIRECT = 307 # ... https://http.dev/307
|
||||
PERMANENT_REDIRECT = 308 # ... https://http.dev/308
|
||||
|
||||
# 4XX - Client Errors:
|
||||
BAD_REQUEST = 400 # ..................... https://http.dev/401
|
||||
UNAUTHORIZED = 401 # .................... https://http.dev/401
|
||||
PAYMENT_REQUIRED = 402 # ................ https://http.dev/402
|
||||
FORBIDDEN = 403 # ....................... https://http.dev/403
|
||||
NOT_FOUND = 404 # ....................... https://http.dev/404
|
||||
METHOD_NOT_ALLOWED = 405 # .............. https://http.dev/405
|
||||
NOT_ACCEPTABLE = 406 # .................. https://http.dev/406
|
||||
PROXY_AUTH_REQUIRED = 407 # ............. https://http.dev/407
|
||||
REQUEST_TIMEOUT = 408 # ................. https://http.dev/408
|
||||
CONFLICT = 409 # ........................ https://http.dev/409
|
||||
GONE = 410 # ............................ https://http.dev/410
|
||||
LENGTH_REQUIRED = 411 # ................. https://http.dev/411
|
||||
PRECONDITION_FAILED = 412 # ............. https://http.dev/412
|
||||
PAYLOAD_TOO_LARGE = 413 # ............... https://http.dev/413
|
||||
URI_TOO_LONG = 414 # .................... https://http.dev/414
|
||||
UNSUPPORTED_MEDIA_TYPE = 415 # .......... https://http.dev/415
|
||||
PAGE_EXPIRED = 419 # .................... https://http.dev/419
|
||||
TOO_MANY_REQUESTS = 429 # ............... https://http.dev/429
|
||||
UNAVAILABLE_FOR_LEGAL_REASONS = 451 # ... https://http.dev/451
|
||||
INVALID_TOKEN = 498 # ................... https://http.dev/498
|
||||
CLIENT_CLOSED_REQUEST = 499 # ........... https://http.dev/499
|
||||
|
||||
# 5XX - Server Errors:
|
||||
INTERNAL_SERVER_ERROR = 500 # ........... https://http.dev/500
|
||||
NOT_IMPLEMENTED = 501 # ................. https://http.dev/501
|
||||
BAD_GATEWAY = 502 # ..................... https://http.dev/502
|
||||
SERVICE_UNAVAILABLE = 503 # ............. https://http.dev/503
|
||||
GATEWAY_TIMEOUT = 504 # ................. https://http.dev/504
|
||||
HTTP_VERSION_NOT_SUPPORTED = 505 # ...... https://http.dev/505
|
||||
VARIANT_ALSO_NEGOTIATES = 506 # ......... https://http.dev/506
|
||||
INSUFFICIENT_STORAGE = 507 # ............ https://http.dev/507
|
||||
LOOP_DETECTED = 508 # ................... https://http.dev/508
|
||||
BANDWIDTH_LIMIT_EXCEEDED = 509 # ........ https://http.dev/509
|
||||
WEB_SERVER_DOWN = 521 # ................. https://http.dev/521
|
||||
ORIGIN_IS_UNREACHABLE = 523 # ........... https://http.dev/523
|
||||
SERVICE_IS_OVERLOADED = 529 # ........... https://http.dev/529
|
||||
NETWORK_READ_TIMEOUT_ERROR = 598 # ...... https://http.dev/598
|
||||
NETWORK_CONNECT_TIMEOUT_ERROR = 599 # ... https://http.dev/599
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@unique
|
||||
class StatusCodes(Enum):
|
||||
|
||||
"""
|
||||
To be used internally within the context of your service. Customize these to match your service.
|
||||
The format is: (SUCCESS_INDICATOR, INTERNAL_NUMERIC_CODE, HTTP_CODE)
|
||||
Example: (True, 1, 200)
|
||||
"""
|
||||
|
||||
# Legacy Codes:
|
||||
OK = (True, 1, HttpCodes.SUCCESS.value)
|
||||
FAILED = (False, 0, HttpCodes.INTERNAL_SERVER_ERROR.value)
|
||||
PARTIAL_SUCCESS = (True, 2, HttpCodes.PARTIAL_CONTENT.value)
|
||||
PARTIAL_FAILURE = (False, 3, HttpCodes.PARTIAL_CONTENT.value)
|
||||
|
||||
# Authentication Codes:
|
||||
LOGGED_IN_SUCCESSFULLY = (True, 200, HttpCodes.SUCCESS.value)
|
||||
LOGIN_FAILED = (False, 201, HttpCodes.UNAUTHORIZED.value)
|
||||
INVALID_SESSION_TOKEN = (False, 202, HttpCodes.UNAUTHORIZED.value)
|
||||
AUTHENTICATION_DETAILS_INCOMPLETE = (False, 203, HttpCodes.BAD_REQUEST.value)
|
||||
|
||||
# Authorization Codes:
|
||||
AUTHORIZED_SUCCESSFULLY = (True, 300, HttpCodes.SUCCESS.value)
|
||||
NOT_ALLOWED = (False, 301, HttpCodes.FORBIDDEN.value)
|
||||
AUTHORIZATION_DETAILS_INCOMPLETE = (False, 302, HttpCodes.BAD_REQUEST.value)
|
||||
AUTHORIZATION_FAILED = (False, 303, HttpCodes.UNAUTHORIZED.value)
|
||||
|
||||
# General failures:
|
||||
DOWN_FOR_MAINTENANCE = (False, 800, HttpCodes.SERVICE_UNAVAILABLE.value)
|
||||
UNKNOWN_ERROR = (False, 801, HttpCodes.INTERNAL_SERVER_ERROR.value)
|
||||
DATA_INCOMPLETE = (False, 802, HttpCodes.BAD_REQUEST.value)
|
||||
HEADERS_INCOMPLETE = (False, 803, HttpCodes.BAD_REQUEST.value)
|
||||
FILES_MISSING = (False, 804, HttpCodes.BAD_REQUEST.value)
|
||||
CLIENT_CLOSED_REQUEST = (False, 805, HttpCodes.CLIENT_CLOSED_REQUEST.value)
|
||||
|
||||
# Validation failure:
|
||||
DATA_VALIDATION_FAILURE = (False, 900, HttpCodes.BAD_REQUEST.value)
|
||||
RATE_LIMIT_EXCEEDED = (False, 901, HttpCodes.TOO_MANY_REQUESTS.value)
|
||||
@@ -1,139 +0,0 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Thursday, 12th Sept., 2024
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To have a structure to the response sent from the API calls.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# System-level activities:
|
||||
import distro
|
||||
import socket
|
||||
import platform
|
||||
|
||||
# For data-modelling:
|
||||
from pydantic import BaseModel, Field
|
||||
from typing import Any, Optional, List, Literal
|
||||
|
||||
# To work with date and time:
|
||||
import datetime
|
||||
|
||||
# My utils:
|
||||
from utils_v2.api.codes import StatusCodes, HttpCodes
|
||||
from utils_v2.date_time import date_time
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# Info for logging that will stay constant during runtime:
|
||||
SERVER_HOSTNAME = str(socket.gethostname())
|
||||
PLATFORM_INFO = platform.uname()
|
||||
HOST_OS = str(distro.name(True))
|
||||
HOST_CPU = f"{PLATFORM_INFO.processor} ({PLATFORM_INFO.machine})"
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** CLASSES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class APILogModel(BaseModel):
|
||||
|
||||
# To identify the machine the code is running on.
|
||||
# DO NOT MODIFY THESE:
|
||||
hostname: str = SERVER_HOSTNAME
|
||||
os: str = HOST_OS
|
||||
cpu: str = HOST_CPU
|
||||
|
||||
# To identify the project and actions:
|
||||
project: Optional[str] = None
|
||||
log: str
|
||||
operation: str
|
||||
apiVer: Optional[str] = None
|
||||
logId: Optional[str] = None
|
||||
logChain: Optional[str] = None
|
||||
|
||||
# Timing metrics:
|
||||
ts: datetime.datetime
|
||||
tat: float
|
||||
cpuTime: float
|
||||
|
||||
# To understand the request that came in:
|
||||
method: Optional[str] = None
|
||||
url: Optional[str] = None
|
||||
route: Optional[str] = None
|
||||
headers: Optional[Any] = None
|
||||
data: Optional[Any] = None
|
||||
files: Optional[Any] = None
|
||||
|
||||
# To understand the output that went out:
|
||||
exception: Optional[Any] = None
|
||||
response: Optional[Any] = None
|
||||
httpCode: Optional[int] = None
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
my_log = APILogModel(
|
||||
log = "internal"
|
||||
)
|
||||
|
||||
print(my_log)
|
||||
@@ -1,133 +0,0 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Thursday, 12th Sept., 2024
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To have a structure to the response sent from the API calls.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# For data-modelling:
|
||||
from pydantic import BaseModel
|
||||
from typing import Any, Optional, List
|
||||
|
||||
# My utils:
|
||||
from utils_v2.api.codes import StatusCodes, HttpCodes
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** CLASSES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class ResponseModel(BaseModel):
|
||||
|
||||
"""
|
||||
A model for how the response should be when developing API endpoints.
|
||||
"""
|
||||
|
||||
# The fields that you want in your response:
|
||||
status_code: StatusCodes
|
||||
message: Optional[str | List] = None
|
||||
data: Optional[Any] = None
|
||||
seconds: Optional[float | int] = None
|
||||
log_id: Optional[str] = None
|
||||
http_code: Optional[HttpCodes] = None
|
||||
api_version: Optional[str] = None
|
||||
|
||||
def for_quart(self):
|
||||
|
||||
"""
|
||||
Call this when you are using either Flask or Quart as your framework.
|
||||
:return: The output as expected by Flask and Quart.
|
||||
"""
|
||||
|
||||
# Construct the basic structure:
|
||||
response_dict = {
|
||||
"status": 1 if self.status_code.value[0] else 0,
|
||||
"code": self.status_code.value[1],
|
||||
"message": self.message or self.status_code.name.replace("_", " ").lower(),
|
||||
"data": self.data,
|
||||
"apiVer": self.api_version
|
||||
}
|
||||
|
||||
# Now add the additional fields:
|
||||
if self.seconds is not None: response_dict["seconds"] = self.seconds
|
||||
if self.log_id is not None: response_dict["logId"] = self.log_id
|
||||
|
||||
# Figure out the HTTP code:
|
||||
response_http_code = self.http_code.value if self.http_code is not None else self.status_code.value[2]
|
||||
|
||||
# Done here:
|
||||
return response_dict, response_http_code
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
my_response = ResponseModel(
|
||||
status_code = StatusCodes.RATE_LIMIT_EXCEEDED
|
||||
)
|
||||
my_response.log_id = "abc123"
|
||||
|
||||
print(my_response.for_quart())
|
||||
Reference in New Issue
Block a user