963 lines
40 KiB
Python
963 lines
40 KiB
Python
"""
|
|
|
|
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.models.behaviour.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())
|