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

(20241129) Started talking to MariaDB for integration.

Browse Source
This commit is contained in:
khushal
2024-11-29 12:58:39 +05:30
parent 4ed028e486
commit aab700b3ed
11 changed files with 902 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/blueprints/mail/oauth_callback.py
+7 -4
View File
@@ -163,8 +163,11 @@ async def mail_callback(
tokens.email = user_profile.data["emailAddress"] if user_profile.success else None
# Save the tokens to the database
tokens_saved = await MailOAuthModel.set_token(
db_conn = current_app.data_mongo,
tokens_saved = await current_app.mail_oauth_model.set_token(
db_conn = current_app.sql_writer,
mongo_conn = current_app.data_mongo,
session_token = inbound_headers["X-Session-Token"],
user_info = kwargs["session_info"],
user_identifier = inbound_data["state"],
token = tokens.model_dump()
)
@@ -174,7 +177,7 @@ async def mail_callback(
# ┛┗┗ ┛┣┛┗┛┛┗┛┗
# ┛
# # Return a response:
# # Return a JSON response:
# return ResponseModel(
# status_code = StatusCodes.OK if tokens_saved else StatusCodes.FAILED,
# http_code = HttpCodes.SUCCESS if tokens_saved else HttpCodes.INTERNAL_SERVER_ERROR,
@@ -184,7 +187,7 @@ async def mail_callback(
# }
# )
# Return a page that shows you the status of your authorization:
# Return an HTML response:
return await render_template(
"/mail/oauth/oauth_success.html" if tokens_saved else "/mail/oauth/oauth_failure.html",
mail_client = mail_client.title()
api/blueprints/mail/oauth_request.py
+4 -5
View File
@@ -66,9 +66,6 @@ from utils_v2.goog.gmail.gmail_client import SCOPES_GMAIL_MAIL_MANAGEMENT
# Common:
from shared import constants
# Behaviour Models:
from models.behaviour.mail.oauth import MailOAuthModel
# Data Models:
from models.data.mail.oauth import (
OAuthMailAuthorizationRequestHeaders,
@@ -178,8 +175,10 @@ async def request_oauth_authorization_url(
# ┛
# Make a user identifier from the session info:
user_identifier = await MailOAuthModel.get_id(
db_conn = current_app.data_mongo,
user_identifier = await current_app.mail_oauth_model.get_user_identifier(
db_conn = current_app.sql_writer,
mongo_conn = current_app.data_mongo,
session_token = inbound_headers["X-Session-Token"],
user_info = kwargs["session_info"],
service_type = "email",
service_client = inbound_data.mailClient,
api/blueprints/tech/__init__.py
View File
api/blueprints/tech/chat_alerts.py
+202
View File
@@ -0,0 +1,202 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Friday, 29th Nov., 2024
OBJECTIVE:
To manage alert sending from one place. You may have various outlets for alerts, like Telegram, WhatsApp, etc.
Add endpoints for them here. Note that these alerts will be for the internal tech admins. It won't be for the
client-facing comms. Create a separate blueprint for that.
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
# My utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
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,
get_session_info,
log_request_to_mongo,
log_chain_to_mongo,
should_not_be_under_maintenance,
only_whitelisted_ips,
limit_rate,
validate_input,
handle_cancelled_request
)
# Data models:
from models.data.tech.alerts import (
ChatAlertRequestHeaders,
ChatAlertRequestData
)
# Common:
from shared import constants
# For asynchronous activities:
import asyncio
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# Related to Quart:
tech_chat_alert_bp = Blueprint("tech_chat_alert", __name__)
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
@tech_chat_alert_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
# ---------------------------------------------------------------------------------------------------------------------
@tech_chat_alert_bp.route("/chat/<source>", methods = ["POST"])
@set_api_version(api_version = "3.0.0")
@read_input(sanitize_headers = True, sanitize_data = True)
@get_session_info(key = "X-Session-Token", session_coro = "get_session")
@should_not_be_under_maintenance(attr_name = "is_under_maintenance")
@validate_input(
header_validator = lambda x: ChatAlertRequestHeaders(**x).model_dump(),
data_validator = lambda x: ChatAlertRequestData(**x)
)
@handle_cancelled_request()
async def user_login(
source: str = None,
inbound_headers: dict = None,
inbound_data: dict | ChatAlertRequestData = None,
inbound_files: dict = None,
**kwargs
):
"""
To relay an alert over some form of chat client like Telegram and WhatsApp.
:param source: The identifier of the source of the alert.
: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.
"""
response = None
# ┳┳┓ ┏┓ •
# ┃┃┃┏┓┏┏┏┓┏┓┏┓ ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┓┏┓┏┓
# ┛ ┗┗ ┛┛┗┻┗┫┗ ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┗┛┛┗
# ┛
# Capture the event's time, and the person who caused the event:
event_dt = date_time.get_current_ist_date_time().strftime("%Y-%m-%d %I:%M:%S %p %Z")
if kwargs.get("session_info"): username = kwargs["session_info"].get("fullName")
else: username = None
# Construct the prefix:
if inbound_data.type == "warning": message_type = "⚠️ WARNING FROM"
elif inbound_data.type == "error": message_type = "🚨 ERROR IN"
else: message_type = "ℹ️ INFO FROM"
message_prefix = f"{message_type} *TCAOFF (Converse)*!\n{event_dt}\n\n"
message_prefix += f"*From:*\n`{username} (via '{source}')`\n\n"
# ┏┳┓ ┓
# ┃ ┏┓┃┏┓┏┓┏┓┏┓┏┳┓
# ┻ ┗ ┗┗ ┗┫┛ ┗┻┛┗┗
# ┛
if inbound_data.chatClient == "telegram":
response = await current_app.http_client.post(
url = current_app.script_data["telegram"]["connectors"]["nexcom"]["url"],
json = {
"appKey": current_app.script_data["telegram"]["connectors"]["nexcom"]["appKey"],
"chatId": inbound_data.chatId or current_app.script_data["telegram"]["chatIds"]["tcaoff"],
"message": message_prefix + inbound_data.message
}
)
# ┳┓
# ┣┫┏┓┏┏┓┏┓┏┓┏┏┓
# ┛┗┗ ┛┣┛┗┛┛┗┛┗
# ┛
# Construct and return the response:
response_json = response.json()
success = True if response_json.get("status", 0) == 1 else False
return ResponseModel(
status_code = StatusCodes.OK if success else StatusCodes.FAILED,
message = response_json.get("message", "unknown failure")
)
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass