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

(20241202) ??

Browse Source
This commit is contained in:
khushal
2024-12-02 18:46:48 +05:30
parent 4d8b40a6c7
commit aed746190d
7 changed files with 684 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/blueprints/mail/sync.py
+198
View File
@@ -0,0 +1,198 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Monday, 2nd Dec., 2024
OBJECTIVE:
To receive requests for synchronising mails from various mail clients to the database. Sync'ing means we pull
the mail from the mail client (like GMail) and store it to our database. The mail is then ready for showing on
the UI at any time.
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, request
# 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,
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
)
# GMail-related utils:
from utils_v2.goog.gmail.gmail_client import SCOPES_GMAIL_MAIL_MANAGEMENT
# Common:
from shared import constants
# Data Models:
from models.data.mail.sync import MailSyncRequestHeaders, MailSyncRequestData
# For asynchronous activities:
import asyncio
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# Related to Quart:
mail_sync_bp = Blueprint("mail_sync", __name__)
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
@mail_sync_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
# ---------------------------------------------------------------------------------------------------------------------
@mail_sync_bp.route("/sync", methods = ["GET"])
@set_api_version(api_version = "1.0.0")
@read_input(sanitize_headers = False, sanitize_data = False)
@get_session_info(key = "X-Session-Token", session_coro = "get_session")
@log_request_to_mongo(
attr_name = "logs_mongo",
project = constants.PROJECT_NAME,
log_type = constants.MODULE_NAME,
operation = "mailOAuthUrlReqApi",
log_input = True,
log_output = True,
sensitive_keys = ["sessionToken"]
)
@log_chain_to_mongo(attr_name = "logs_mongo")
@should_not_be_under_maintenance(attr_name = "is_under_maintenance")
@validate_input(
header_validator = lambda x: MailSyncRequestHeaders(**x).model_dump(),
data_validator = lambda x: MailSyncRequestData(**x)
)
@handle_cancelled_request()
async def sync_mail(
inbound_headers: dict | MailSyncRequestHeaders = None,
inbound_data: dict | MailSyncRequestData = None,
inbound_files: dict = None,
**kwargs
):
"""
Use this when the user wants to pull old mails from some mail client (like GMail) and save it to the database for
ready access on the UI.
: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 the session token is invalid/expired:
if kwargs.get("session_info") is None:
return ResponseModel(
status_code = StatusCodes.FAILED,
http_code = HttpCodes.UNAUTHORIZED
)
# Start by assuming failure:
auth_url = None
# ┏┓ ┏┳┓ ┓
# ┃┓┏┓╋ ┃ ┏┓┃┏┏┓┏┓
# ┗┛┗ ┗ ┻ ┗┛┛┗┗ ┛┗
# Make a user identifier from the session info:
user_tokens = await current_app.mail_oauth_model.get_token(
mongo_conn = current_app.data_mongo,
user = kwargs["session_info"]
)
print("MAIL TOKEN(S):", json.to_string(user_tokens))
# ┳┓
# ┣┫┏┓┏┏┓┏┓┏┓┏┏┓
# ┛┗┗ ┛┣┛┗┛┛┗┛┗
# ┛
# Done here:
return ResponseModel(status_code = StatusCodes.OK)
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
models/behaviour/mail/oauth_v2.py
+312
View File
@@ -0,0 +1,312 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Wednesday, 27th Nov., 2024
OBJECTIVE:
To define the interaction between the UI layer and the database connectivity in one place. Here we shall handle
all the activities for OAuth2.0 authorization requests for all the users of our service.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# My async utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
from utils_v2.database.async_mysql_v2 import AsyncMySQL
from utils_v2.database.async_mongo_v2 import AsyncMongo
# Base model:
from models.behaviour.base import BaseModel
# To work with MongoDB:
from bson import ObjectId
# To work with datatypes:
from typing import Literal
# To make deep-copies:
import copy
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class MailOAuthModel(BaseModel):
AUTH_COLLECTION = "_authTokens"
async def get_user_identifier(
self,
db_conn: AsyncMySQL,
mongo_conn: AsyncMongo,
user_info: dict,
service_client: Literal["gmail"],
auth_type: Literal["oauth"],
session_token: str = None
) -> ObjectId:
"""
Stores params from the session info and gives an identifier to use in the authorization URL. Use this when the
user requests an authorization URL to link your service to another service (like GMail).
:param db_conn: The database connection (MariaDB) to use to perform the action.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param user_info: The dictionary that has the user's session information.
:param service_client: The name of the company or brand that is providing this service that is being integrated.
:param auth_type: To identify the type of authentication being done here. This could indicate simple password
authentication, more advance OAuth2.0 authentication, etc.
:param session_token: The session token of the user who requested this service.
:return: An ObjectId to later store the granted tokens.
"""
# Note down the timestamp at which this event occurred:
request_ts = date_time.get_current_utc_date_time(as_string = False)
# Get the identifier from the database:
inserted_id = await mongo_conn.insert_one(
collection = MailOAuthModel.AUTH_COLLECTION,
document = {
"version": "-1.0.1",
"serviceType": "email",
"client": service_client,
"authType": auth_type,
"user": user_info,
"token": None,
"firstRefreshTs": None,
"lastRefreshTs": None,
"lastRequestTs": request_ts
}
)
return inserted_id
# # Get the identifier from the database:
# mongo_json = await mongo_conn.find_one_and_update(
# collection = MailOAuthModel.AUTH_COLLECTION,
# filter = {
# "serviceType": "email",
# "client": service_client,
# "authType": auth_type,
# "user": user_info,
# },
# update = {
# "$set": {
# "lastRequestTs": request_ts
# },
# "$setOnInsert": {
# "version": "1.0.0",
# "serviceType": "email",
# "client": service_client,
# "authType": auth_type,
# "user": user_info,
# "token": None,
# "firstRefreshTs": None,
# "lastRefreshTs": None,
# "firstRequestTs": request_ts,
# }
# },
# projection = {
# "_id": True
# },
# upsert = True,
# return_updated = True
# )
#
# # Tell MariaDB that an authorization request was initiated:
# db_json = {}
# if mongo_json is not None:
# db_json = await self.call_procedure(
# db_conn = db_conn,
# proc_name = "entity_integration_save",
# proc_args = (
# user_info["entityId"], # ............................................ 'p_entity_id'
# service_client, # ................................................... 'p_provider'
# "Auth Requested", # ................................................. 'p_current_status'
# "Auth URL Generated", # ............................................. 'p_last_action'
# None, # ............................................................. 'p_display_name'
# None, # ............................................................. 'p_display_picture'
# str(mongo_json["_id"]), # ........................................... 'p_token_id'
# json.to_string(python_data = {"email": None}, no_space = True), # ... 'p_notes'
# user_info["userId"] # ............................................... 'p_created_by'
# ),
# session_token = session_token
# )
#
# # Done here:
# return mongo_json["_id"] if mongo_json and db_json.get("status") == 1 else None
async def set_token(
self,
db_conn: AsyncMySQL,
mongo_conn: AsyncMongo,
user_identifier: ObjectId | str,
token: dict,
session_token: str = None
) -> bool:
"""
This method is to be called when the end user authorizes your service to connect to his third-party account. For
example, when the end user allows you to access his GMail account. USE THIS FOR UPDATING (REFRESHING) TOKENS
ALSO.
:param db_conn: The database connection (MariaDB) to use to perform the action.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param user_identifier: The identifier granted by the 'get_user_identifier' method.
:param token: The token granted by the third-party service.
:param session_token: The session token of the user who requested this service.
:return: True if saved, False if failed.
"""
# Start by assuming failure:
token_saved = False
# Note down the timestamp at which this event occurred:
request_ts = date_time.get_current_utc_date_time(as_string = False)
# Save the token to MongoDB:
mongo_json = await mongo_conn.find_one_and_update(
collection = MailOAuthModel.AUTH_COLLECTION,
filter = {"_id": ObjectId(user_identifier)},
update = {
"$set": {
"token": token,
"lastRefreshTs": request_ts,
},
"$setOnInsert": {
"firstRefreshTs": request_ts,
}
},
projection = {"token": False},
return_updated = True,
upsert = False
)
# Tell MariaDB that the token was saved:
if mongo_json is not None:
db_json = await self.call_procedure(
db_conn = db_conn,
proc_name = "entity_integration_save",
proc_args = (
mongo_json["user"]["entityId"], # ............................................. 'p_entity_id'
mongo_json["client"], # ....................................................... 'p_provider'
"Auth Granted", # ............................................................. 'p_current_status'
"Set Token", # ................................................................ 'p_last_action'
None, # ....................................................................... 'p_display_name'
None, # ....................................................................... 'p_display_picture'
user_identifier, # ............................................................ 'p_token_id'
json.to_string(python_data = {"email": token["email"]}, no_space = True), # ... 'p_notes'
mongo_json["user"]["userId"] # ................................................ 'p_created_by'
),
session_token = session_token
)
if db_json["status"] == 1: token_saved = True
# Done here:
return token_saved
async def get_token(
self,
mongo_conn: AsyncMongo,
user_identifier: ObjectId | str = None,
**kwargs
) -> dict | None:
"""
To retrieve stored tokens from the database.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param user_identifier: The identifier granted by the 'get_user_identifier' method.
:param kwargs: Any set of key-value pairs to build custom search criteria. This could be things like the user
info, the client, the type of authentication used, or even the kind of service.
:return: The retrieved record that has the token, and information about the service and client if found, else
None when there is no matching record.
"""
# Build the filter:
filter_json = {k: v for k, v in kwargs.items()}
if user_identifier: filter_json["_id"] = ObjectId(user_identifier)
# If there is no search criteria, we exit with failure:
if not filter_json: return None
# If there is some filtering possible,
# we fetch and return the token:
return await mongo_conn.find_one(
collection = self.AUTH_COLLECTION,
filter = filter_json,
projection = {
"_id": True,
"serviceType": True,
"authType": True,
"client": True,
"token": True
}
)
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
models/data/mail/sync.py
+133
View File
@@ -0,0 +1,133 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Wednesday, 27th Nov., 2024.
OBJECTIVE:
To provide the structure for the request and response of the APIs that will be used to request OAuth2.0
authorization for mail services.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# For making data behaviour_models:
from pydantic import BaseModel, Field, field_validator
from typing import Optional, Literal
# My utils:
from utils_v2.string import regex
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# RegEx Patterns:
REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$"
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
class OAuthMailAuthorizationRequestHeaders(BaseModel):
sessionToken: str = Field(
description = "the session token of the user who is requesting the service",
pattern = REGEX_SESSION_TOKEN,
frozen = True,
alias = "X-Session-Token"
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "allow"
def model_dump(self, *args, **kwargs):
return super().model_dump(*args, by_alias = True, **kwargs)
# ---------------------------------------------------------------------------------------------------------------------
class OAuthMailAuthorizationRequestData(BaseModel):
mailClient: Literal["gmail"] = Field(
description = "the e-mail provider like 'gmail'",
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("mailClient", mode = "before")
def to_lowercase(cls, value):
if isinstance(value, str): value = value.strip().lower()
return value
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
utils_v2/database/async_mongo_v2.py
+1 -1
View File
@@ -959,7 +959,7 @@ class AsyncMongo(AsyncMongoBase):
:param requests: The array of requests (operations) to be performed.
:param session: The session if you need to do this in a transaction.
:param raise_exception: Whether, or not, you want to raise an exception when something fails.
:return:
:return: The no of operations done.
"""
# Ensure you are connected:
utils_v2/date_time/date_time.py
+2
View File
@@ -67,6 +67,8 @@ DATE_TIME_FORMATS = (
"%d/%b",
"%d%m%Y",
"%Y%m%d",
"%Y/%m/%d",
"%Y-%m-%d",
"%Y-%m-%d %H:%M:%S"
)
utils_v2/goog/base.py
+13
View File
@@ -148,6 +148,19 @@ class AsyncGoogleBase:
def debug_everything(self):
self._debug_only_errors = False
# ┏┓ •
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
# ┛
@property
def client_id(self):
return self._client_id
@property
def client_secret(self):
return self._client_secret
# ┏┓┏┓ ┓ ┏┓ ┏┓
# ┃┃┣┫┓┏╋┣┓ ┏┛ ┃┫
# ┗┛┛┗┗┻┗┛┗ ┗━•┗┛
utils_v2/goog/gmail/gmail_client.py
+25 -18
View File
@@ -201,7 +201,7 @@ class AsyncGMailClient(AsyncGoogleBase):
if api_response.httpCode in [200]:
api_response.success = True
api_json = await api_response.get_json()
api_response.data = {label.pop("name"): label for label in api_json.get("labels", [])}
api_response.data = {label["name"]: label for label in api_json.get("labels", [])}
# Done here:
return api_response
@@ -417,7 +417,7 @@ class AsyncGMailClient(AsyncGoogleBase):
async def __list_messages_on_page(
self,
tokens: GoogleAuthTokens,
count: int = 100,
max_count: int = 100,
query: str = None,
label_ids: List[str] | str = None,
include_spam_and_trash: bool = False,
@@ -434,7 +434,7 @@ class AsyncGMailClient(AsyncGoogleBase):
1. https://developers.google.com/gmail/api/reference/rest/v1/users.messages/list
2. https://developers.google.com/gmail/api/reference/rest/v1/users.messages#Message
:param tokens: The object that holds the access token to the service.
:param count: The no. of messages to fetch.
:param max_count: The no. of messages to fetch.
:param query: Any query filter that is supported by GMail.
:param label_ids: The list of labels' ids that the mails must have on them.
:param include_spam_and_trash: Whether, or not, you would like to include mails categorized as spam and trash.
@@ -453,7 +453,7 @@ class AsyncGMailClient(AsyncGoogleBase):
# Build the needed params:
params_json = {
"maxResults": count,
"maxResults": max_count,
"includeSpamTrash": include_spam_and_trash
}
if query: params_json["q"] = query
@@ -461,7 +461,7 @@ class AsyncGMailClient(AsyncGoogleBase):
if label_ids: params_json["labelIds"] = label_ids if isinstance(label_ids, list) else [label_ids]
# Make the API call:
if not self._debug_only_errors: self._printer("Listing Messages for Page.", user_id, count, next_page_token)
if not self._debug_only_errors: self._printer("Listing Messages for Page.", user_id, max_count, next_page_token)
api_response = await self.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/messages",
headers = {"Authorization": f"Bearer {tokens.accessToken}"},
@@ -473,7 +473,7 @@ class AsyncGMailClient(AsyncGoogleBase):
api_response.success = True
api_json = await api_response.get_json()
api_response.data = {
"messages": {m.pop("id"): m for m in api_json.get("messages", [])},
"messages": {m["id"]: m for m in api_json.get("messages", [])},
"nextPageToken": api_json.get("nextPageToken"),
"resultSizeEstimate": api_json["resultSizeEstimate"],
}
@@ -484,7 +484,7 @@ class AsyncGMailClient(AsyncGoogleBase):
async def list_messages(
self,
tokens: GoogleAuthTokens,
count: int = 100,
max_count: int = 100,
query: str = None,
label_ids: List[str] | str = None,
include_spam_and_trash: bool = False,
@@ -498,7 +498,7 @@ class AsyncGMailClient(AsyncGoogleBase):
1. https://developers.google.com/gmail/api/reference/rest/v1/users.messages/list
2. https://developers.google.com/gmail/api/reference/rest/v1/users.messages#Message
:param tokens: The object that holds the access token to the service.
:param count: The no. of messages to fetch.
:param max_count: The no. of messages to fetch.
:param query: Any query filter that is supported by GMail.
:param label_ids: The list of labels' ids that the mails must have on them.
:param include_spam_and_trash: Whether, or not, you would like to include mails categorized as spam and trash.
@@ -532,8 +532,8 @@ class AsyncGMailClient(AsyncGoogleBase):
# Let's figure out how many times we'll have to loop through the process to retrieve the target no. of
# messages. Google allows you to fetch info about at most 500 messages in one go.
max_per_call = 500 # ... because Google allows at most 500 entries in one call.
iterations_needed = int(math.ceil(count / max_per_call))
last_iteration_count = count - int((max_per_call * (iterations_needed - 1)))
iterations_needed = int(math.ceil(max_count / max_per_call))
last_iteration_count = max_count - int((max_per_call * (iterations_needed - 1)))
# Run the loop those many times:
results_size_estimate = 0
@@ -543,12 +543,12 @@ class AsyncGMailClient(AsyncGoogleBase):
if iterations_needed > 1:
if iteration_no < (iterations_needed - 1): iteration_count = max_per_call
else: iteration_count = last_iteration_count
else: iteration_count = count
else: iteration_count = max_count
# Retrieve the messages for this page:
iteration_response = await self.__list_messages_on_page(
tokens = tokens,
count = iteration_count,
max_count = iteration_count,
query = query,
label_ids = label_ids,
include_spam_and_trash = include_spam_and_trash,
@@ -957,13 +957,20 @@ if __name__ == "__main__":
# print(my_mail.get_raw_message(as_base64 = False))
# Test some feature:
# response = await my_gmail.list_messages(
# tokens = test_tokens,
# )
response = await my_gmail.get_message(
tokens = test_tokens,
message_id = "1936bfbfc912b86f"
auth_url = await my_gmail.get_authorization_url(
state = "1234567890",
scopes = SCOPES_GMAIL_MAIL_MANAGEMENT,
approval_prompt = "force",
user_email = "pskhushal@gmail.com"
)
print("AUTH URL:", auth_url)
response = await my_gmail.list_labels(
tokens = test_tokens,
)
# response = await my_gmail.get_message(
# tokens = test_tokens,
# message_id = "1936bfbfc912b86f"
# )
print("SUCCESS:", response.success)
print("SUMMARY:", response.to_markdown())
print("\n\n---\n\n")