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

(20241126) GMail v2 started with better data handling and feedback.

Browse Source
This commit is contained in:
khushal
2024-11-26 20:36:50 +05:30
parent db80153cb4
commit 629922431d
10 changed files with 1245 additions and 333 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/goog/base.py
+375
View File
@@ -0,0 +1,375 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Tuesday, 26th Nov., 2024
OBJECTIVE:
To provide a base class for common behaviour of Google's APIs.
REFERENCES:
1. Quickstart: https://developers.google.com/gmail/api/quickstart/python
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.date_time import date_time
from utils_v2.oauth.services.goog import GoogleOAuth
from utils_v2.goog.models.data.api_call import GoogleApiResponse
from utils_v2.mail import mail_parser
# Related to Google:
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
# To make API calls:
import httpx
# For asynchronous activities:
import asyncio
# To work with date and time:
import datetime
# For working with datatypes:
from typing import Dict, Literal, List, Any
# For debugging:
from icecream import IceCreamDebugger
import inspect
# For computational help:
import math
# For base64 encoding:
import base64
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AsyncGoogleBase:
def __init__(
self,
service_name: str,
client_id: str,
client_secret: str,
http_client: httpx.AsyncClient,
debug = True,
debug_prefix = "GMail | ",
debug_only_errors = True
):
"""
To initialize any Google API from one base class. The client's id and secret are available in the file
downloaded form https://console.cloud.google.com/apis/credentials (do not forget to select your app).
:param service_name: A string to identify this service.
:param client_id: From the OAuth JSON downloaded from
:param http_client: An asynchronous HTTP client to make API calls.
:param debug: Whether, or not, you would like to show debugging messages on the terminal.
:param debug_prefix: The prefix string to identify the debugging messages.
:param debug_only_errors: Whether you would like to show all debugging messages or just error messages.
"""
# Prepare the debugging utility:
self._debug_prefix = debug_prefix
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
if not debug: self._printer.disable()
self._debug_only_errors = debug_only_errors
# Accept the input configuration:
self._service_name = service_name
self._client_id = client_id
self._client_secret = client_secret
self._http_client = http_client
def enable_debug(self):
self._printer.enable()
def disable_debug(self):
self._printer.disable()
def debug_only_errors(self):
self._debug_only_errors = True
def debug_everything(self):
self._debug_only_errors = False
# ┏┓ ┳┓ ┓•
# ┣ ┏┓┏┓┏┓┏┓ ┃┃┏┓┏┏┓┏┫┓┏┓┏┓
# ┗┛┛ ┛ ┗┛┛ ┻┛┗ ┗┗┛┗┻┗┛┗┗┫
# ┛
@staticmethod
async def get_error_message(api_response: GoogleApiResponse):
try: return (await api_response.get_json())["error"]["message"]
except: return api_response.response.reason_phrase
# ┏┓┏┓┳ ┏┓ ┓┓•
# ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓
# ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫
# ┛
async def get(
self,
url: str,
headers: dict = None,
params: dict = None
) -> GoogleApiResponse:
"""
To call an API using the GET method.
:param url: The URL to call.
:param headers: The headers to pass.
:param params: The params to send in the query string itself.
:return: A structured response that includes the raw response, the exception (if any), and so on.
"""
# Prepare the structure of the response:
api_response = GoogleApiResponse(
serviceName = self._service_name,
action = inspect.stack()[1].function,
url = url,
method = "GET"
)
try:
# Make the API call:
response = await self._http_client.get(
url = url,
headers = headers,
params = params
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = await self.get_error_message(api_response)
# If something goes wrong:
except Exception as exception:
api_response.exception = exception
api_response.message = str(exception)
self._printer(exception, api_response.url, api_response.method, headers, params)
# Done here:
return api_response
async def post(
self,
url: str,
headers: dict = None,
json: dict = None,
data: dict = None
) -> GoogleApiResponse:
"""
To call an API using the POST method.
:param url: The URL to call.
:param headers: The headers to pass.
:param json: The params to send in the JSON body.
:param data: The params to send in the form-data in the body.
:return: A structured response that includes the raw response, the exception (if any), and so on.
"""
# Prepare the structure of the response:
api_response = GoogleApiResponse(
serviceName = self._service_name,
action = inspect.stack()[1].function,
url = url,
method = "POST"
)
try:
# Make the API call:
response = await self._http_client.post(
url = url,
headers = headers,
json = json,
data = data
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = await self.get_error_message(api_response)
# If something goes wrong:
except Exception as exception:
api_response.exception = exception
api_response.message = str(exception)
self._printer(exception, api_response.url, api_response.method, headers, json, data)
# Done here:
return api_response
async def put(
self,
url: str,
headers: dict = None,
json: dict = None,
data: dict = None
) -> GoogleApiResponse:
"""
To call an API using the PUT method.
:param url: The URL to call.
:param headers: The headers to pass.
:param json: The params to send in the JSON body.
:param data: The params to send in the form-data in the body.
:return: A structured response that includes the raw response, the exception (if any), and so on.
"""
# Prepare the structure of the response:
api_response = GoogleApiResponse(
serviceName = self._service_name,
action = inspect.stack()[1].function,
url = url,
method = "PUT"
)
try:
# Make the API call:
response = await self._http_client.put(
url = url,
headers = headers,
json = json,
data = data
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = await self.get_error_message(api_response)
# If something goes wrong:
except Exception as exception:
api_response.exception = exception
api_response.message = str(exception)
self._printer(exception, api_response.url, api_response.method, headers, json, data)
# Done here:
return api_response
async def delete(
self,
url: str,
headers: dict = None
) -> GoogleApiResponse:
"""
To call an API using the DELETE method.
:param url: The URL to call.
:param headers: The headers to pass.
:return: A structured response that includes the raw response, the exception (if any), and so on.
"""
# Prepare the structure of the response:
api_response = GoogleApiResponse(
serviceName = self._service_name,
action = inspect.stack()[1].function,
url = url,
method = "DELETE"
)
try:
# Make the API call:
response = await self._http_client.delete(
url = url,
headers = headers
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = await self.get_error_message(api_response)
# If something goes wrong:
except Exception as exception:
api_response.exception = exception
api_response.message = str(exception)
self._printer(exception, api_response.url, api_response.method, headers)
# Done here:
return api_response
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
utils_v2/goog/gmail/gmail.py
+130 -60
View File
@@ -45,6 +45,7 @@ from utils_v2.string import json
from utils_v2.string import regex from utils_v2.string import regex
from utils_v2.date_time import date_time from utils_v2.date_time import date_time
from utils_v2.oauth.services.goog import GoogleOAuth from utils_v2.oauth.services.goog import GoogleOAuth
from utils_v2.mail import mail_parser
# Related to Google: # Related to Google:
from google.auth.transport.requests import Request from google.auth.transport.requests import Request
@@ -69,6 +70,9 @@ from icecream import IceCreamDebugger
# For computational help: # For computational help:
import math import math
# For base64 encoding:
import base64
# ***************************************************************************************************************** # *****************************************************************************************************************
# ***** **** # ***** ****
@@ -679,63 +683,26 @@ class AsyncGMailClient:
# Done here: # Done here:
return messages return messages
@staticmethod
def format_message(
raw_message: Dict[str, Any],
consider_timezone = date_time.TIMEZONE_UTC,
) -> Dict[str, Any] | None:
message_ts = date_time.parse_date_time(raw_message["internalDate"], timezone = consider_timezone)
message_ts = date_time.to_timezone(message_ts, date_time.TIMEZONE_UTC)
message_headers = {h["name"]: h["value"] for h in raw_message["payload"]["headers"]}
message = {
"ts": message_ts,
"messageId": raw_message["id"],
"threadId": raw_message["threadId"],
"labels": raw_message["labelIds"],
"summary": raw_message["snippet"],
"headers": message_headers,
"from": regex.find_first(text = message_headers["From"], pattern = regex.REGEX_EMAIL_ID),
"to": regex.find_first(text = message_headers["To"], pattern = regex.REGEX_EMAIL_ID),
"cc": [
regex.find_first(
text = cc_mail.strip(),
pattern = regex.REGEX_EMAIL_ID
) for cc_mail in cc_mails.split(",")
] if (cc_mails := message_headers.get("Cc")) else None,
"bcc": [
regex.find_first(
text = bcc_mail.strip(),
pattern = regex.REGEX_EMAIL_ID
) for bcc_mail in bcc_mails.split(",")
] if (bcc_mails := message_headers.get("Bcc")) else None,
"subject": message_headers["Subject"]
}
# Done here:
return message
async def get_message( async def get_message(
self, self,
message_id: str, message_id: str,
return_raw = False, return_raw = False,
consider_timezone = date_time.TIMEZONE_UTC,
raise_exception: bool = False raise_exception: bool = False
) -> Dict[str, Any] | None: ) -> Dict[str, Any] | str | None:
""" """
To get one message of this user. The message will be identified by its id. To get one message of this user. The message will be identified by its id. Note that, if you choose to return
the raw message, the message body will be compliant with RFC 5322 and RFC 2045 (among others).
DOCUMENTATION: DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.messages/list 1. https://developers.google.com/gmail/api/reference/rest/v1/users.messages/get
2. https://developers.google.com/gmail/api/reference/rest/v1/users.messages#Message 2. https://developers.google.com/gmail/api/reference/rest/v1/users.messages#Message
2. https://developers.google.com/gmail/api/reference/rest/v1/users.messages#Message.MessagePart 2. https://developers.google.com/gmail/api/reference/rest/v1/Format
:param message_id: The id that Google assigned to the message. :param message_id: The id that Google assigned to the message.
:param return_raw: Whether you want the raw message or the formatted message. :param return_raw: Whether you want the raw message or the formatted message.
:param consider_timezone: The timestamp given by Google doesn't have timezone information. Use this parameter to
control what timezone the timestamp is interpreted as.
:param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be :param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be
suppressed. suppressed.
:return: The list of labels. :return: The message either parsed as a JSON, or as a raw text body. If the API call fails, the response will be
a null value.
""" """
# Start by assuming failure: # Start by assuming failure:
@@ -750,21 +717,25 @@ class AsyncGMailClient:
if not self._debug_only_errors: self._printer("Getting One Message.", self.__user_email) if not self._debug_only_errors: self._printer("Getting One Message.", self.__user_email)
api_response = await self.__http_client.get( api_response = await self.__http_client.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/messages/{message_id}", url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/messages/{message_id}",
headers = {"Authorization": f"Bearer {self.__credentials.token}"} headers = {"Authorization": f"Bearer {self.__credentials.token}"},
params = {"format": "raw"}
) )
# If the API call failed: # If the API call failed:
if api_response.status_code not in [200]: return message if api_response.status_code not in [200]: return message
# Return the raw message if asked: # Extract the raw message and respond based on the request:
api_json = api_response.json() api_json = api_response.json()
if return_raw: return api_json message = base64.urlsafe_b64decode(api_json["raw"])
if return_raw: message = message.decode()
# Else we extract and format the response: else:
message = self.format_message( message = mail_parser.parse(message)
raw_message = api_json, message["labels"] = api_json["labelIds"]
consider_timezone = consider_timezone message["messageId"] = api_json["id"]
) message["threadId"] = api_json["threadId"]
message["historyId"] = api_json["historyId"]
message["snippet"] = api_json["snippet"]
message["sizeEstimate"] = api_json["sizeEstimate"]
# In case something goes wrong along the way: # In case something goes wrong along the way:
except Exception as exception: except Exception as exception:
@@ -775,6 +746,97 @@ class AsyncGMailClient:
# Done here: # Done here:
return message return message
async def modify_messages(
self,
message_ids: List[str] | str,
add_label_ids: List[str] | str = None,
remove_label_ids: List[str] | str = None,
raise_exception: bool = False
) -> bool:
"""
To add or remove labels from messages.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.messages/batchModify
:param message_ids: One or more message ids (assigned by Google).
:param add_label_ids: One or more label ids (not the display name of the label).
:param remove_label_ids: One or more label ids (not the display name of the label).
:param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be
suppressed.
:return: True if the operation succeeded, else False.
"""
# Start by assuming failure:
success = False
try:
# Standard token-refresh check:
await self.__ensure_token()
# Build the needed JSON:
body_json = {"ids": message_ids if isinstance(message_ids, list) else [message_ids]}
if add_label_ids: body_json["addLabelIds"] = add_label_ids if isinstance(add_label_ids, list) else [add_label_ids]
if remove_label_ids: body_json["removeLabelIds"] = remove_label_ids if isinstance(remove_label_ids, list) else [remove_label_ids]
# Make the API call:
if not self._debug_only_errors: self._printer("Modifying Message(s).", self.__user_email)
api_response = await self.__http_client.post(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/messages/batchModify",
headers = {"Authorization": f"Bearer {self.__credentials.token}"},
json = body_json
)
# Check if our request was successful:
if api_response.status_code in [200, 204]: success = True
# In case something goes wrong along the way:
except Exception as exception:
if raise_exception: raise
self._printer(exception)
success = False
# Done here:
return success
async def delete_messages(
self,
message_ids: List[str] | str,
raise_exception: bool = False
) -> bool:
# Start by assuming failure:
success = False
try:
# Standard token-refresh check:
await self.__ensure_token()
# Make the API call:
if not self._debug_only_errors: self._printer("Deleting Message(s).", self.__user_email)
api_response = await self.__http_client.post(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/messages/batchDelete",
headers = {"Authorization": f"Bearer {self.__credentials.token}"},
json = {"ids": message_ids if isinstance(message_ids, list) else [message_ids]}
)
print("HTTP CODE:", api_response.status_code)
try: print("HTTP JSON:", json.to_string(api_response.json()))
except Exception as e: print("HTTP JSON:", e)
# Check if our request was successful:
if api_response.status_code in [200, 204]: success = True
# In case something goes wrong along the way:
except Exception as exception:
if raise_exception: raise
self._printer(exception)
success = False
# Done here:
return success
# ***************************************************************************************************************** # *****************************************************************************************************************
# ***** **** # ***** ****
@@ -854,13 +916,21 @@ if __name__ == "__main__":
# print(json.to_string(await my_gmail.delete_label(label_id = "Label_6"))) # print(json.to_string(await my_gmail.delete_label(label_id = "Label_6")))
# print(json.to_string(await my_gmail.list_messages(count = 10))) # print(json.to_string(await my_gmail.list_messages(count = 10)))
print(json.to_string( # print(json.to_string(
await my_gmail.get_message( # await my_gmail.get_message(
message_id = "1935e92672e67f22", # message_id = "19367930033154ca",
return_raw = True # return_raw = False
), # ),
default = str) # default = str)
) # )
# print(json.to_string(await my_gmail.modify_messages(
# message_ids = ["19367930033154ca"],
# add_label_ids = ["Label_5"],
# remove_label_ids = ["Label_3"]
# )))
print(json.to_string(await my_gmail.delete_messages(message_ids = "19367930033154ca")))
asyncio.run(main()) asyncio.run(main())
utils_v2/goog/gmail/gmail_v2.py
+251
View File
@@ -0,0 +1,251 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Monday, 25th Nov., 2024
OBJECTIVE:
To manage e-mails in a GMail account.
REFERENCES:
1. Quickstart: https://developers.google.com/gmail/api/quickstart/python
2. Labels: https://developers.google.com/gmail/api/guides/labels
3. Messages: https://developers.google.com/gmail/api/reference/rest/v1/users.messages
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.date_time import date_time
from utils_v2.mail import mail_parser
# My Google utils:
from utils_v2.oauth.services.goog import GoogleOAuth
from utils_v2.goog.base import AsyncGoogleBase
from utils_v2.goog.models.data.auth_tokens import GoogleAuthTokens
from utils_v2.goog.models.data.api_call import GoogleApiResponse
# Related to Google:
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
# To make API calls:
import httpx
# For asynchronous activities:
import asyncio
# To work with date and time:
import datetime
# For working with datatypes:
from typing import Dict, Literal, List, Any
# For debugging:
from icecream import IceCreamDebugger
# For computational help:
import math
# For base64 encoding:
import base64
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AsyncGMailClient(AsyncGoogleBase):
async def get_user_profile(
self,
tokens: GoogleAuthTokens,
user_id: str = "me",
) -> GoogleApiResponse:
# Ensure that the tokens are valid:
await tokens.arefresh(
http_client = self._http_client,
client_id = self._client_id,
client_secret = self._client_secret,
force_refresh = False
)
# Make the API call:
if not self._debug_only_errors: self._printer("Getting User Profile.")
api_response = await self.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/profile",
headers = {"Authorization": f"Bearer {tokens.accessToken}"}
)
# If the call was successful:
if api_response.httpCode in [200]:
api_response.data = await api_response.get_json()
# Done here:
return api_response
# ┓ ┓ ┓
# ┃ ┏┓┣┓┏┓┃┏
# ┗┛┗┻┗┛┗ ┗┛
async def list_labels(
self,
tokens: GoogleAuthTokens,
user_id: str = "me"
) -> GoogleApiResponse:
"""
To get the list of labels of this user.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/list
2. https://developers.google.com/gmail/api/reference/rest/v1/users.labels#Label
:param tokens: The object that holds the access token to the service.
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself.
:return: A structured response where the list of labels will be in the 'data' variable.
"""
# Ensure that the tokens are valid:
await tokens.arefresh(
http_client = self._http_client,
client_id = self._client_id,
client_secret = self._client_secret,
force_refresh = False
)
# Make the API call:
if not self._debug_only_errors: self._printer("Listing All Labels.", user_id)
api_response = await self.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/labels",
headers = {"Authorization": f"Bearer {tokens.accessToken}"}
)
# If the call was successful:
if api_response.httpCode in [200]:
api_json = await api_response.get_json()
api_response.data = {label.pop("name"): label for label in api_json.get("labels", [])}
# Done here:
return api_response
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
import dateparser
from utils_v2.goog.models.data.auth_tokens import GoogleAuthTokens
# Create an HTTP client:
test_client = httpx.AsyncClient(
limits = httpx.Limits(
max_connections = 100, # ............ Maximum number of connections allowed in the pool.
max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive.
),
timeout = httpx.Timeout(
connect = 2.5, # ... Shorter connection timeout.
read = 2.5, # ...... Like what EasyEcom gives.
write = 10.0, # .... Time to wait for sending data.
pool = 120.0 # ..... Time to wait for a free connection from the pool.
)
)
secrets_file = r"../../../creds/google_tcaoff_test_oauth_20241125.json"
secrets_dict = json.from_file(secrets_file)
tokens = {
"accessToken": "ya29.a0AeDClZAYoo85BXRId_n-hwo_amKshzi46c33GaJcsZZvGB7A7OGU2RFYcWBM_BleNBfAFUSJP2NHAvmd7Nsp_U5Kg68hXSy0iO99PNTm3pvKrJSzbkA-rXsVLsCnBIfPUMyNt2nOOVJmGwm17DNN0jAELkm1fPNTju7SZzmuaCgYKAZwSARMSFQHGX2Mis8TZui2rZT1gKySVds-N0w0175",
"refreshToken": "1//0gnqzjMf9YT19CgYIARAAGBASNgF-L9Ir3rcY37nGrV45XyOUBRllEH7Txui7T1JbwevlmDoNw7PuMu149cCWQSwsScuKaZusUQ",
"expiresIn": 3539,
"expiresAt": dateparser.parse("2024-11-25 10:40:40.833699+00:00"),
"scopes": [
"https://www.googleapis.com/auth/gmail.labels",
"https://www.googleapis.com/auth/gmail.modify"
]
}
async def main():
my_gmail = AsyncGMailClient(
service_name = "gmail",
client_id = secrets_dict["web"]["client_id"],
client_secret = secrets_dict["web"]["client_secret"],
http_client = test_client,
debug = True,
debug_prefix = "GMail (M) | ",
debug_only_errors = False
)
test_tokens = GoogleAuthTokens(**tokens)
response = await my_gmail.get_user_profile(tokens = test_tokens)
print("RESPONSE:", response)
asyncio.run(main())
utils_v2/goog/models/__init__.py
View File
utils_v2/goog/models/data/__init__.py
View File
utils_v2/goog/models/data/api_call.py
+131
View File
@@ -0,0 +1,131 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Wednesday, 30th Oct., 2024.
OBJECTIVE:
To provide a data model for describing the API response from Google's APIs.
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, model_validator
from typing import Optional, Literal, Union, Dict, List, Any
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
class GoogleApiResponse(BaseModel):
serviceName: str = Field(frozen = True, default = None)
action: str = Field(frozen = True, default = None)
url: str = Field(frozen = True)
method: str = Field(frozen = True)
response: Any = None
httpCode: int = None
success: bool = False
message: str = None
data: Any = None
exception: Any = None
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ┏┓ ┏┓
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
def to_markdown(self):
if self.exception: message = "❌ *GOOGLE API EXCEPTION:* ❌\n\n"
else: message = "*PROVIDER API RESPONSE:*\n\n"
message += f"*SERVICE:*\n`{self.serviceName}`\n\n"
message += f"*ACTION:*\n`{self.action}`\n\n"
message += f"*URL:*\n`{self.url}`\n\n"
message += f"*METHOD:*\n`{self.method}`\n\n"
message += f"*RESPONSE:*\n`{self.response}`\n\n"
message += f"*EXCEPTION:*\n`{self.exception.__class__.__name__}: {str(self.exception)}`\n\n"
return message
async def get_json(self):
try: return self.response.json()
except: return {}
async def get_content(self):
try: return self.response.content
except: return b""
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
utils_v2/goog/models/data/auth_tokens.py
+216
View File
@@ -0,0 +1,216 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Wednesday, 30th Oct., 2024.
OBJECTIVE:
To provide a data model for describing the tokens to be used for Google's APIs.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
import httpx
sys.path.append(".")
sys.path.append("..")
# For making data behaviour_models:
from pydantic import BaseModel, Field, field_validator, model_validator, AwareDatetime
from typing import Optional, Literal, Union, Dict, List, Any
# Related to Google:
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.date_time import date_time
# To work with date and time:
import datetime
import dateparser
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
class GoogleAuthTokens(BaseModel):
accessToken: str = Field(description = "the main 'bearer' token")
refreshToken: str = Field(description = "token to be used to refresh the access token")
expiresAt: AwareDatetime = Field(description = "the time (utc) at which the token will expire")
scopes: List[str] = Field(description = "the list of permissions", default = [])
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "ignore"
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("expiresAt", mode = "before")
def parse_dates(cls, value):
if not isinstance(value, datetime.datetime):
parsed = date_time.parse_date_time(value, date_formats = ["%Y%m%d", "%Y-%m-%d"])
value = parsed if isinstance(parsed, datetime.datetime) else dateparser.parse(value)
if isinstance(value, datetime.datetime): value = date_time.to_timezone(value, date_time.TIMEZONE_UTC)
return value
# ┏┓ •
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
# ┛
@property
def expired(self):
return True if date_time.get_current_utc_date_time() >= self.expiresAt else False
@property
def ttl(self):
return (self.expiresAt - date_time.get_current_utc_date_time()).total_seconds()
# ┏┓ ┏┓
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
def refresh(
self,
client_id: str,
client_secret: str,
force_refresh: bool = False
) -> bool:
"""
Synchronously (blocking) refreshes the existing access tokens in place.
:param client_id: The id of the client (OAuth JSON) for which these tokens were granted.
:param client_secret: The secret of the client (OAuth JSON) for which these tokens were granted.
:param force_refresh: Whether you want to refresh the token even if it hasn't expired.
:return: True if refreshed, else False.
"""
# Start by assuming failure:
success = False
try:
# Go ahead only if either the token has expired,
# or the user has asked to forcefully refresh the tokens:
if self.expired or force_refresh:
# Create the credentials:
credentials = Credentials.from_authorized_user_info(
info = {
"client_id": client_id,
"client_secret": client_secret,
"refresh_token": self.refreshToken,
"expires_at": self.expiresAt
}
)
# Request a refresh:
credentials.refresh(Request())
# Note down the new credentials:
if credentials.token != self.accessToken:
success = True
self.accessToken = credentials.token
self.refreshToken = credentials.refresh_token
self.expiresAt = date_time.as_if_timezone(credentials.expiry, timezone = date_time.TIMEZONE_UTC)
# In case something goes wrong:
except Exception as exception:
success = False
# Done here:
return success
async def arefresh(
self,
http_client: httpx.AsyncClient,
client_id: str,
client_secret: str,
force_refresh: bool = False
) -> bool:
"""
Asynchronously refreshes the existing access tokens in place.
:param http_client: The HTTP client to use to make the refresh request.
:param client_id: The id of the client (OAuth JSON) for which these tokens were granted.
:param client_secret: The secret of the client (OAuth JSON) for which these tokens were granted.
:param force_refresh: Whether you want to refresh the token even if it hasn't expired.
:return: True if refreshed, else False.
"""
# Currently we don't really know how to refresh tokens through low-level API calls,
# so we will pass on the intent to the regular, synchronous function.
return self.refresh(
client_id = client_id,
client_secret = client_secret,
force_refresh = force_refresh
)
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
utils_v2/mail/async_mail.py → utils_v2/mail/async_smpt.py
+5 -5
View File
@@ -97,7 +97,7 @@ from typing import List
# ***************************************************************************************************************** # *****************************************************************************************************************
class MailMessage: class SMTPMessage:
def __init__( def __init__(
self, self,
@@ -212,7 +212,7 @@ class MailMessage:
# --------------------------------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------------------------------
class AsyncMailClient: class AsyncSMTPClient:
# Constants: # Constants:
SMTP_TLS_PORT = 587 SMTP_TLS_PORT = 587
@@ -336,7 +336,7 @@ class AsyncMailClient:
self.__smtp = None self.__smtp = None
return True return True
async def send(self, mail: MailMessage): async def send(self, mail: SMTPMessage):
""" """
Send out the mail. Send out the mail.
@@ -395,14 +395,14 @@ if __name__ == "__main__":
async def test(): async def test():
mail_client = AsyncMailClient( mail_client = AsyncSMTPClient(
email = "sender@gmail.com", email = "sender@gmail.com",
password = "zcaf nmqy ncfz fave", password = "zcaf nmqy ncfz fave",
server = "smtp.gmail.com", server = "smtp.gmail.com",
rate_limiters = None rate_limiters = None
) )
my_mail = MailMessage( my_mail = SMTPMessage(
to_email = "orangebhopli@gmail.coms", to_email = "orangebhopli@gmail.coms",
subject = "Bhopli is the best!", subject = "Bhopli is the best!",
cc_emails = None, cc_emails = None,
utils_v2/mail/mail_message.py
-268
View File
@@ -1,268 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Tuesday, 16th Jul, 2024
OBJECTIVE:
To be able to send out mails from code.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# For working with mails:
import aiosmtplib
from email.mime.multipart import MIMEMultipart
from email.mime.image import MIMEImage
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
# My utils:
from utils import rate_limit_utils
# Common:
from shared.statuses import StatusCodes
# For random strings:
import string
import random
# For system-level activities:
import os
# For working with files in RAM:
import io
# For debugging:
from icecream import IceCreamDebugger
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class MailMessage:
def __init__(self, to_email, subject):
"""
Create an instance of the message that you would like to send.
:param to_email: The EMail ID of th recipient.
:param subject: The subject of the mail.
"""
self.message = MIMEMultipart()
self.message["To"] = to_email
self.message["Subject"] = subject
def add_text(self, text):
"""
Add plain-text to the mail body.
:param text: The text to add to the mail body.
:return: None.
"""
self.message.attach(MIMEText(text, "plain"))
def add_html(self, html_text):
"""
Add HTML text to the mail body.
:param html_text: The HTML text to add to the mail body.
:return: None.
"""
self.message.attach(MIMEText(html_text, "html"))
def add_inline_image(self, image_file, content_id = None):
"""
Add an inline image to the body of the mail.
NOTE: This is NOT the same as sending an image as an attachment.
:param image_file: The image data to attach to the mail body.
:param content_id: Inline images are inserted via HTML bocks. This field identifies the image resource. If not
specified, I will generate a random string. You may write a custom value here if you know what you are
doing. For most use cases, please ignore this field.
:return: None.
"""
# Read the image as bytes:
image_bytes = None
if type(image_file) is str:
with open(image_file, "rb") as opened_image_file:
image_bytes = opened_image_file.read()
if type(image_file) is io.BytesIO:
image_file.seek(0)
image_bytes = image_file.getvalue()
# Declare the part to be attached to the multipart message:
if image_bytes is not None:
# Create the HTML block if the image pointer is blank:
if content_id is None:
content_id = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(8))
self.add_html(f"""
<html>
<body>
<p><img src="cid:{content_id}"></p>
</body>
</html>
""")
# Then add the image:
image_part = MIMEImage(image_bytes)
image_part.add_header("Content-ID", f"<{content_id}>")
self.message.attach(image_part)
def add_attachment(self, attachment_file, file_name = None):
# Declare the part to be attached to the multipart message:
part = MIMEBase("application", "octet-stream")
# If the attachment is a file stored in the local disk:
if type(attachment_file) is str:
file_name = file_name or os.path.split(attachment_file)[-1]
with open(attachment_file, "rb") as attachment:
part.set_payload(attachment.read())
# If the file is held in RAM:
if type(attachment_file) is io.BytesIO():
attachment_file.seek(0)
part.set_payload(attachment_file.read())
# Encode and attach the file:
encoders.encode_base64(part)
part.add_header(
"Content-Disposition",
f"attachment; filename= {file_name}",
)
self.message.attach(part)
def get_message(self):
return self.message
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
import asyncio
from utils import json_utils
from utils_v2.mail.async_mail import AsyncMailClient
async def test():
rate_lim = rate_limit_utils.TokenBucket(
rate_limit = 1,
seconds = 60.0,
)
mail_client = AsyncMailClient(
email = "sender@gmail.com",
password = "secret_password",
server = "smtp.gmail.com",
rate_limiters = rate_lim
)
my_mail = MailMessage(
to_email = "recipient@gmail.com",
subject = "Bhopli is the best!"
)
my_mail.add_html(
"""
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Sample HTML String</title>
<style>
.heading {
color: #ff9025;
}
.sub-heading {
color: #000000;
}
</style>
</head>
<body>
<h1 class="heading">Hello, Bhopli!</h1>
<h2 class="sub-heading">Bhopli is the best, most well-behaved cat in the known universe.</h2>
</body>
</html>
"""
)
my_mail.add_text("This is how you should pet her 👇")
my_mail.add_inline_image(r"/path/to/image/cat_petting.png")
my_mail.add_attachment(r"/path/to/file/sample_label.pdf")
await mail_client.login()
result = await mail_client.send(my_mail)
print("MAIL RESULT:", json_utils.to_json_string(result))
await mail_client.logout()
asyncio.run(test())
utils_v2/mail/mail_parser.py
+137
View File
@@ -0,0 +1,137 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Tuesday, 26th Nov., 2024
OBJECTIVE:
To parse raw mail bodies and give a structure that is suitable for storing in No-SQL databases like MongoDB. The
raw mail's text is expected to be compliant with standard defined in RFC 5322, RFC 2045, and maybe a few more.
REFERENCES:
1. GitHub: https://github.com/SpamScope/mail-parser
2. RFC 5322: https://datatracker.ietf.org/doc/html/rfc5322
3. RFC 2045: https://datatracker.ietf.org/doc/html/rfc2045
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.date_time import date_time
# To work with mails:
import mailparser
# To work with datatypes:
from typing import Any, Dict
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
def parse(raw_mail: str | bytes) -> Dict[str, Any]:
"""
To parse the raw mail text to a usable JSON that can even be stored on a No-SQL database like MongoDB.
DOCUMENTATION:
1. GitHub: https://github.com/SpamScope/mail-parser
2. RFC 5322: https://datatracker.ietf.org/doc/html/rfc5322
3. RFC 2045: https://datatracker.ietf.org/doc/html/rfc2045
:param raw_mail: The raw mail body that adheres to RFC 5322 and RFC 2045 (among others).
:return: The parsed JSON format (dict) of the mail.
"""
# Parse the raw format:
if isinstance(raw_mail, str): parsed_mail = mailparser.parse_from_string(raw_mail)
else: parsed_mail = mailparser.parse_from_bytes(raw_mail)
# Format the attachments:
message_attachments = [
{
"filename": attachment["filename"],
"type": attachment["mail_content_type"],
"cid": regex.find_first(text = attachment["content-id"], pattern = r"(?<=<).*(?=>)"),
"rawCid": attachment["content-id"],
"contentDisposition": (cd := attachment["content-disposition"]),
"isInline": True if cd.lower().find("inline") >= 0 else False,
"charset": attachment["charset"],
"contentTransferEncoding": attachment["content_transfer_encoding"],
"payload": attachment["payload"]
} for attachment in parsed_mail.attachments
]
# Put everything together:
return {
"ts": date_time.to_timezone(parsed_mail.date, timezone = date_time.TIMEZONE_UTC),
"headers": parsed_mail.headers,
"from": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers["From"]],
"to": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers["To"]],
"cc": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers.get("Cc", [])],
"bcc": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers.get("Bcc", [])],
"text": parsed_mail.text_plain,
"html": parsed_mail.text_html,
"attachments": message_attachments,
}
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass