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

Resetting utils subtree.

Browse Source
This commit is contained in:
khushal
2024-12-04 18:52:06 +05:30
parent 9b496587de
commit 8d415bd5f6
134 changed files with 0 additions and 101800 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/goog/gmail/gmail_client.py
-962
View File
@@ -1,962 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Monday, 25th Nov., 2024
OBJECTIVE:
To manage e-mails in a GMail account.
REFERENCES:
1. GMail Quickstart: https://developers.google.com/gmail/api/quickstart/python
2. GMail Labels: https://developers.google.com/gmail/api/guides/labels
3. GMail Messages: https://developers.google.com/gmail/api/reference/rest/v1/users.messages
4. People Profile: https://developers.google.com/people/api/rest/v1/people/get
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.goog.base import AsyncGoogleBase
from utils_v2.goog.models.data.auth_tokens import GoogleAuthTokens
from utils_v2.goog.models.data.api_call import GoogleApiResponse
from utils_v2.goog.gmail.gmail_message import GMailMessage
# 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 ***
# ***** ****
# *****************************************************************************************************************
# Google Scopes:
SCOPES_GMAIL_MAIL_MANAGEMENT = [
r"https://www.googleapis.com/auth/gmail.modify",
r"https://www.googleapis.com/auth/gmail.labels",
# r"profile",
r"https://www.googleapis.com/auth/userinfo.profile"
]
SCOPES_GMAIL_FULL = [
r"https://mail.google.com/",
# r"profile",
r"https://www.googleapis.com/auth/userinfo.profile"
]
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AsyncGMailClient(AsyncGoogleBase):
async def get_user_profile(
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/getProfile
2. https://developers.google.com/people/api/rest/v1/people/get
3. https://developers.google.com/people/api/rest/v1/people#Person
: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 or "me".
: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 GMail API call:
if not self._debug_only_errors: self._printer("Getting User Profile.")
gmail_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 gmail_api_response.httpCode in [200]:
gmail_api_response.success = True
gmail_api_response.data = await gmail_api_response.get_json()
gmail_api_response.data["displayName"] = None
gmail_api_response.data["displayPictureUrl"] = None
# Make the People API call:
if not self._debug_only_errors: self._printer("Getting User Profile.")
people_api_response = await self.get(
url = f"https://people.googleapis.com/v1/people/me?personFields=names,photos,birthdays,phoneNumbers,genders,emailAddresses,addresses",
headers = {"Authorization": f"Bearer {tokens.accessToken}"}
)
# If the call was successful:
if people_api_response.httpCode in [200]:
people_api_response.success = True
people_api_response.data = await people_api_response.get_json()
for item in people_api_response.data.get("names", []):
if item["metadata"]["primary"]:
gmail_api_response.data["displayName"] = item.get("displayName")
for item in people_api_response.data.get("photos", []):
if item["metadata"]["primary"]:
gmail_api_response.data["displayPictureUrl"] = item.get("url")
# Done here:
return gmail_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 or "me".
: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_response.success = True
api_json = await api_response.get_json()
api_response.data = {label["name"]: label for label in api_json.get("labels", [])}
# Done here:
return api_response
async def get_label(
self,
tokens: GoogleAuthTokens,
label_id: str,
user_id: str = "me"
) -> GoogleApiResponse:
"""
To get one label of this user. the label will be identified by its id.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/get
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 label_id: The id that Google assigned to the label.
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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("Getting One Label.", user_id)
api_response = await self.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/labels/{label_id}",
headers = {"Authorization": f"Bearer {tokens.accessToken}"}
)
# If the call was successful:
if api_response.httpCode in [200]:
api_response.success = True
api_response.data = await api_response.get_json()
# Done here:
return api_response
async def create_label(
self,
tokens: GoogleAuthTokens,
label_name: str,
label_visibility: Literal["labelShow", "labelShowIfUnread", "labelHide"] = "labelShow",
message_visibility: Literal["show", "hide"] = "show",
label_text_color: str = "#434343",
label_background_color: str = "#000000",
user_id: str = "me"
) -> GoogleApiResponse:
"""
Create one label for the user. Doesn't apply it to any mail, just creates it.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/create
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 label_name: The display name of the label.
:param label_visibility: Whether, or not, you would like to show the label in the web UI.
:param message_visibility: Whether, or not, you would like to show messages with this label in the web UI.
:param label_text_color: The colour of the text of the label.
:param label_background_color: The colour of the background/tag of the label.
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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("Creating One Label.", user_id)
api_response = await self.post(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/labels",
headers = {"Authorization": f"Bearer {tokens.accessToken}"},
json = {
"name": label_name,
"messageListVisibility": "show" if message_visibility else "hide",
"labelListVisibility": "labelShow" if label_visibility else "labelHide",
"color": {
"textColor": label_text_color.lower(),
"backgroundColor": label_background_color.lower()
}
}
)
# If the call was successful:
if api_response.httpCode in [200]:
api_response.success = True
api_response.data = await api_response.get_json()
# Done here:
return api_response
async def update_label(
self,
tokens: GoogleAuthTokens,
label_id: str,
label_name: str = None,
label_visibility: Literal["labelShow", "labelShowIfUnread", "labelHide"] = None,
message_visibility: Literal["show", "hide"] = None,
label_text_color: str = None,
label_background_color: str = None,
user_id: str = "me"
) -> GoogleApiResponse:
"""
Updates one label for the user.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/update
2. https://developers.google.com/gmail/api/reference/rest/v1/users.labels#Label
NOTE: Both or none of the colours must be updated. For this reason, a simple default will be chosen for the
other if only one is provided.
:param tokens: The object that holds the access token to the service.
:param label_id: The id that Google assigned to the label.
:param label_name: The display name of the label.
:param label_visibility: Whether, or not, you would like to show the label in the web UI.
:param message_visibility: Whether, or not, you would like to show messages with this label in the web UI.
:param label_text_color: The colour of the text of the label.
:param label_background_color: The colour of the background/tag of the label.
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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
)
# Format the JSON body:
json_body = {}
if label_name: json_body["name"] = label_name
if label_visibility: json_body["labelListVisibility"] = label_visibility
if message_visibility: json_body["messageListVisibility"] = message_visibility
if label_text_color or label_background_color:
json_body["color"] = {
"textColor": (label_text_color or "#434343").lower(),
"backgroundColor": (label_background_color or "#000000").lower()
}
# Make the API call:
if not self._debug_only_errors: self._printer("Updating One Label.", user_id)
api_response = await self.put(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/labels/{label_id}",
headers = {"Authorization": f"Bearer {tokens.accessToken}"},
json = json_body
)
# If the call was successful:
if api_response.httpCode in [200]:
api_response.success = True
api_response.data = await api_response.get_json()
# Done here:
return api_response
async def delete_label(
self,
tokens: GoogleAuthTokens,
label_id: str,
user_id: str = "me"
) -> GoogleApiResponse:
"""
To delete one label of this user. the label will be identified by its id.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.labels/delete
:param tokens: The object that holds the access token to the service.
:param label_id: The id that Google assigned to the label.
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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("Deleting One Label.", user_id)
api_response = await self.delete(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/labels/{label_id}",
headers = {"Authorization": f"Bearer {tokens.accessToken}"}
)
# If the call was successful:
if api_response.httpCode in [200, 204]:
api_response.success = True
# Done here:
return api_response
# ┳┳┓
# ┃┃┃┏┓┏┏┏┓┏┓┏┓┏
# ┛ ┗┗ ┛┛┗┻┗┫┗ ┛
# ┛
async def __list_messages_on_page(
self,
tokens: GoogleAuthTokens,
max_count: int = 100,
query: str = None,
label_ids: List[str] | str = None,
include_spam_and_trash: bool = False,
next_page_token: str = None,
user_id: str = "me"
) -> GoogleApiResponse:
"""
To enlist messages on one page. Google allows at most 500 results on one page. This method respects that
pagination limit and returns only what Google gives. This method should be used internally by the class and the
class should expose another method that calls this one in loop to get any arbitrary no. of messages as the user
desires.
DOCUMENTATION:
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 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.
:param next_page_token: The token to fetch the next set of results.
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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
)
# Build the needed params:
params_json = {
"maxResults": max_count,
"includeSpamTrash": include_spam_and_trash
}
if query: params_json["q"] = query
if next_page_token: params_json["pageToken"] = next_page_token
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, 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}"},
params = params_json
)
# If the call was successful:
if api_response.httpCode in [200]:
api_response.success = True
api_json = await api_response.get_json()
api_response.data = {
"messages": {m["id"]: m for m in api_json.get("messages", [])},
"nextPageToken": api_json.get("nextPageToken"),
"resultSizeEstimate": api_json["resultSizeEstimate"],
}
# Done here:
return api_response
async def list_messages(
self,
tokens: GoogleAuthTokens,
max_count: int = 100,
query: str = None,
label_ids: List[str] | str = None,
include_spam_and_trash: bool = False,
next_page_token: str = None,
user_id: str = "me"
) -> GoogleApiResponse:
"""
To enlist mail messages from a user's account.
DOCUMENTATION:
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 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.
:param next_page_token: The token to fetch the next set of results.
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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
)
# We create a variable that will hold the results.
# We must supply the URL, Method and a few other params here due to the custom looping functionality:
all_messages = GoogleApiResponse(
serviceName = self._service_name,
action = inspect.stack()[0].function,
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/messages",
method = "GET",
data = {
"messages": {},
"nextPageToken": None,
"resultSizeEstimate": 0
}
)
# 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(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
for iteration_no in range(iterations_needed):
# Figure out the count for this page:
if iterations_needed > 1:
if iteration_no < (iterations_needed - 1): iteration_count = max_per_call
else: iteration_count = last_iteration_count
else: iteration_count = max_count
# Retrieve the messages for this page:
iteration_response = await self.__list_messages_on_page(
tokens = tokens,
max_count = iteration_count,
query = query,
label_ids = label_ids,
include_spam_and_trash = include_spam_and_trash,
next_page_token = next_page_token
)
# Check if no data was received:
if (
iteration_response.data is None or
not iteration_response.data.get("messages")
): break
# Now that we know that messages were received:
for k, v in iteration_response.data["messages"].items(): all_messages.data["messages"][k] = v
results_size_estimate += iteration_response.data["resultSizeEstimate"]
# Also copy the API call params:
all_messages.httpCode = iteration_response.httpCode
all_messages.message = iteration_response.message
all_messages.success = iteration_response.success
# If there is no next page after this, we break out of the loop:
next_page_token = iteration_response.data["nextPageToken"]
if next_page_token is None: break
# Format the final response:
all_messages.data["nextPageToken"] = next_page_token
all_messages.data["resultSizeEstimate"] = results_size_estimate
# Done here:
return all_messages
async def get_message(
self,
tokens: GoogleAuthTokens,
message_id: str,
return_raw: bool = False,
user_id: str = "me"
) -> GoogleApiResponse:
"""
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:
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/Format
:param tokens: The object that holds the access token to the service.
: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 user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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("Getting One Message.", user_id)
api_response = await self.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/messages/{message_id}",
headers = {"Authorization": f"Bearer {tokens.accessToken}"},
params = {"format": "raw"}
)
# If the call was successful:
if api_response.httpCode in [200]:
api_response.success = True
api_json = await api_response.get_json()
raw_message = base64.urlsafe_b64decode(api_json["raw"]).decode()
if return_raw: api_response.data = raw_message
else:
parsed_message = mail_parser.parse(raw_message)
parsed_message["labels"] = api_json["labelIds"]
parsed_message["messageId"] = api_json["id"]
parsed_message["threadId"] = api_json["threadId"]
parsed_message["historyId"] = api_json["historyId"]
parsed_message["snippet"] = api_json["snippet"]
parsed_message["sizeEstimate"] = api_json["sizeEstimate"]
api_response.data = parsed_message
# Done here:
return api_response
async def modify_messages(
self,
tokens: GoogleAuthTokens,
message_ids: List[str] | str,
add_label_ids: List[str] | str = None,
remove_label_ids: List[str] | str = None,
user_id: str = "me"
) -> GoogleApiResponse:
"""
To add or remove labels from one or more messages.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.messages/batchModify
:param tokens: The object that holds the access token to the service.
: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 user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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
)
# 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).", user_id)
api_response = await self.post(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/messages/batchModify",
headers = {"Authorization": f"Bearer {tokens.accessToken}"},
json = body_json
)
# If the call was successful:
if api_response.httpCode in [200, 204]:
api_response.success = True
# Done here:
return api_response
async def delete_messages(
self,
tokens: GoogleAuthTokens,
message_ids: List[str] | str,
user_id: str = "me"
) -> GoogleApiResponse:
"""
To PERMANENTLY delete one or more messages.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.messages/batchDelete
:param tokens: The object that holds the access token to the service.
:param message_ids: One or more message ids (assigned by Google).
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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("Deleting Message(s).", user_id)
api_response = await self.post(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/messages/batchDelete",
headers = {"Authorization": f"Bearer {tokens.accessToken}"},
json = {"ids": message_ids if isinstance(message_ids, list) else [message_ids]}
)
# If the call was successful:
if api_response.httpCode in [200, 204]:
api_response.success = True
# Done here:
return api_response
async def trash_message(
self,
tokens: GoogleAuthTokens,
message_id: str,
user_id: str = "me"
) -> GoogleApiResponse:
"""
To move one message to trash.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.messages/trash
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 message_id: The id of the message (assigned by Google).
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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("Trashing One Message.", user_id)
api_response = await self.post(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/messages/{message_id}/trash",
headers = {"Authorization": f"Bearer {tokens.accessToken}"},
)
# If the call was successful:
if api_response.httpCode in [200, 204]:
api_response.success = True
api_response.data = await api_response.get_json()
# Done here:
return api_response
async def untrash_message(
self,
tokens: GoogleAuthTokens,
message_id: str,
user_id: str = "me"
) -> GoogleApiResponse:
"""
To move one message to trash.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.messages/trash
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 message_id: The id of the message (assigned by Google).
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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("Un-Trashing One Message.", user_id)
api_response = await self.post(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/messages/{message_id}/untrash",
headers = {"Authorization": f"Bearer {tokens.accessToken}"},
)
# If the call was successful:
if api_response.httpCode in [200]:
api_response.success = True
api_response.data = await api_response.get_json()
# Done here:
return api_response
async def send_message(
self,
tokens: GoogleAuthTokens,
message: GMailMessage,
thread_id: str = None,
user_id: str = "me"
) -> GoogleApiResponse:
"""
To send one message.
NOTE: If you want to apply a custom label to your outgoing mails, this API endpoint doesn't allow you to do that
in one go. Instead, you should note down the 'id' field from a successful response and use the 'modify_messages'
method of this class to immediately apply that label to the mail in a separate call.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users.messages/send
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 message: The object that has the content of the message to be sent.
:param thread_id: Replies to an existing mail if the correct thread-id is specified. If not specified, a new
mail with a new thread-id is created.
:param user_id: The id of the user for which this service must be run. Typically, the e-mail id itself or "me".
: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
)
# Construct the JSON body:
json_body = {"raw": message.get_raw_message(as_base64 = True)}
if thread_id: json_body["threadId"] = thread_id
# Make the API call:
if not self._debug_only_errors: self._printer("Sending One Message.", user_id)
api_response = await self.post(
url = f"https://gmail.googleapis.com/gmail/v1/users/{user_id}/messages/send",
headers = {
"Authorization": f"Bearer {tokens.accessToken}",
"Content-Type": "message/rfc822"
},
json = json_body
)
# If the call was successful:
if api_response.httpCode in [200]:
api_response.success = True
api_response.data = await api_response.get_json()
# Done here:
return api_response
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
# 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.
)
)
# Read the secrets that give you access to the app:
secrets_file = r"../../../creds/goog/app/google_tcaoff_test_oauth_20241125.json"
secrets_dict = json.from_file(secrets_file)
async def main():
# Create an instance of the client:
my_gmail = AsyncGMailClient(
service_name = "gmail",
oauth_json = secrets_dict,
http_client = test_client,
redirect_url = r"https://api.thecaoffice.com/converse/mail/callback/gmail",
debug = True,
debug_prefix = "GMail (M) | ",
debug_only_errors = False
)
# Request Auth:
print("AUTH URL:", await my_gmail.get_authorization_url(
scopes = SCOPES_GMAIL_MAIL_MANAGEMENT,
state = "Bhopli",
approval_prompt = "force"
))
# Get tokens from callback:
test_tokens = await my_gmail.get_authorization_tokens(
scopes = SCOPES_GMAIL_MAIL_MANAGEMENT,
redirect_url = input("Paste the redirect URL here: ")
)
print("TOKENS:", test_tokens)
# Test some feature:
response = await my_gmail.get_user_profile(tokens = test_tokens)
print("SUCCESS:", response.success)
print("SUMMARY:", response.to_markdown())
print("\n\n---\n\n")
print("DATA:", json.to_string(response.data, default = str))
if not response.success:
print("\n\n---\n\n")
print("FULL RESPONSE JSON:", json.to_string(await response.get_json()))
asyncio.run(main())