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

(20241127) Mail sending (with replies) tested :)

Browse Source
This commit is contained in:
khushal
2024-11-27 13:54:29 +05:30
parent 759b5a05be
commit 535f998272
12 changed files with 1768 additions and 1403 deletions
Download Patch File Download Diff File Expand all files Collapse all files
creds/google_converse_test_oauth_20241122.json → creds/goog/app/google_converse_test_oauth_20241122.json
View File
creds/google_tcaoff_test_oauth_20241125.json → creds/goog/app/google_tcaoff_test_oauth_20241125.json
View File
creds/goog/user/test_user_tokens_20241125.json
View File
data/images/cat_petting.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 37 KiB

data/pdf/sample_label.pdf
+481
View File
File diff suppressed because one or more lines are too long
utils_v2/goog/base.py
+4 -2
View File
@@ -232,7 +232,8 @@ class AsyncGoogleBase:
url: str, url: str,
headers: dict = None, headers: dict = None,
json: dict = None, json: dict = None,
data: dict = None data: dict = None,
content: str | bytes = None
) -> GoogleApiResponse: ) -> GoogleApiResponse:
""" """
@@ -259,7 +260,8 @@ class AsyncGoogleBase:
url = url, url = url,
headers = headers, headers = headers,
json = json, json = json,
data = data data = data,
content = content
) )
# Note down the results: # Note down the results:
utils_v2/goog/gmail/gmail.py
-936
View File
@@ -1,936 +0,0 @@
"""
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.oauth.services.goog import GoogleOAuth
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
# 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:
def __init__(
self,
credentials: Credentials,
http_client: httpx.AsyncClient,
debug = True,
debug_prefix = "GMail | ",
debug_only_errors = True
):
# 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.__http_client = http_client
self.__service = None
self.__user_email = None
self.__credentials = credentials
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
# ┏┓ ┏┓ ┏┓ ┏•
# ┗┓┏┓╋┓┏┏┓ ┣╋ ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗ ┗┗┻┣┛ ┗┻ ┗┛┗┛┛┗┛┗┗┫
# ┛ ┛
async def initialize(self):
"""
Call this once when the instance is created.
:return: True if successful, else False.
"""
return await self.__build_service()
async def __build_service(self) -> bool:
"""
Build a service object that can be used to perform various activities.
:return: True if successful, else False.
"""
try:
# Build the service object:
self.__service = build(
serviceName = "gmail",
version = "v1",
credentials = self.__credentials,
cache_discovery = False
)
# Note down the e-mail address of the user:
profile = self.__service.users().getProfile(userId = "me").execute()
self.__user_email = profile.get("emailAddress")
# Done here:
return True
# In case something goes wrong:
except Exception as exception:
self._printer(exception)
return False
async def set_credentials(
self,
credentials: Credentials
) -> bool:
"""
To update the credentials. Needed for the times when the access token gets refreshed.
:param credentials: Google's custom 'Credentials' object that describes the OAuth-based access details.
:return: True if successful, else False.
"""
self.__credentials = credentials
return await self.__build_service()
async def __ensure_token(self) -> None:
"""
Checks if the access token has expired and refreshes if needed.
:return: None.
"""
if await GoogleOAuth.credentials_have_expired(self.__credentials):
if not self._debug_only_errors: self._printer("Refreshing Token.", self.__user_email)
self.__credentials.refresh(Request())
# ┏┓ •
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
# ┛
@property
async def user_email(self):
return self.__user_email
@property
async def user_profile(self):
return self.__service.users().getProfile(userId = "me").execute()
# ┓ ┓ ┓
# ┃ ┏┓┣┓┏┓┃┏
# ┗┛┗┻┗┛┗ ┗┛
async def list_labels(
self,
raise_exception: bool = False
) -> Dict[str, dict] | None:
"""
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 raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be
suppressed.
:return: The list of labels.
"""
# Start by assuming failure:
labels = None
try:
# Standard token-refresh check:
await self.__ensure_token()
# Make the API call:
if not self._debug_only_errors: self._printer("Listing All Labels.", self.__user_email)
api_response = await self.__http_client.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/labels",
headers = {"Authorization": f"Bearer {self.__credentials.token}"}
)
# If the API call failed:
if api_response.status_code not in [200]: return labels
# Else we format the response:
labels = {label.pop("name"): label for label in api_response.json().get("labels", [])}
# In case something goes wrong along the way:
except Exception as exception:
if raise_exception: raise
self._printer(exception)
labels = None
# Done here:
return labels
async def get_label(
self,
label_id: str,
raise_exception: bool = False
) -> dict | None:
"""
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 label_id: The id that Google assigned to the label.
:param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be
suppressed.
:return: The list of labels.
"""
# Start by assuming failure:
label = None
try:
# Standard token-refresh check:
await self.__ensure_token()
# Make the API call:
if not self._debug_only_errors: self._printer("Getting One Label.", self.__user_email)
api_response = await self.__http_client.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/labels/{label_id}",
headers = {"Authorization": f"Bearer {self.__credentials.token}"}
)
# If the API call failed:
if api_response.status_code not in [200]: return label
# Else we extract the response:
label = api_response.json()
# In case something goes wrong along the way:
except Exception as exception:
if raise_exception: raise
self._printer(exception)
label = None
# Done here:
return label
async def create_label(
self,
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",
raise_exception: bool = False
) -> bool:
"""
Create one label for the user.
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 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 raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be
suppressed.
:return: True if the label was created, else False.
"""
# 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("Creating One Label.", self.__user_email)
api_response = await self.__http_client.post(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/labels",
headers = {
"Authorization": f"Bearer {self.__credentials.token}"
},
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 API call failed:
if api_response.status_code not in [200]: return success
# Else we note down the success:
if api_response.json().get("id") is not None: 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 update_label(
self,
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,
raise_exception: bool = False
) -> bool:
"""
Updates one label for the user.
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
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 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 raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be
suppressed.
:return: True if the label was created, else False.
"""
# Start by assuming failure:
success = False
try:
# Standard token-refresh check:
await self.__ensure_token()
# 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()
}
# If no value was given to update:
if not json_body: return success
# Make the API call:
if not self._debug_only_errors: self._printer("Updating One Label.", self.__user_email)
api_response = await self.__http_client.put(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/labels/{label_id}",
headers = {"Authorization": f"Bearer {self.__credentials.token}"},
json = json_body
)
# If the API call failed:
if api_response.status_code not in [200]: return success
# Else we note down the success:
if api_response.json().get("id") is not None: 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_label(
self,
label_id: str,
raise_exception: bool = False
) -> bool:
"""
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/get
2. https://developers.google.com/gmail/api/reference/rest/v1/users.labels#Label
:param label_id: The id that Google assigned to the label.
:param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be
suppressed.
:return: The list of labels.
"""
# 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 One Label.", self.__user_email)
api_response = await self.__http_client.delete(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/labels/{label_id}",
headers = {"Authorization": f"Bearer {self.__credentials.token}"}
)
# If the API call failed:
if api_response.status_code not in [200, 204]: return success
# Else we extract the response:
else: 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 __list_messages_on_page(
self,
count: int = 100,
query: str = None,
label_ids: List[str] | str = None,
include_spam_and_trash: bool = False,
next_page_token: str = None,
raise_exception: bool = False
) -> Dict[str, Any] | None:
"""
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 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 raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be
suppressed.
:return: The messages that matched the given conditions if the call was successful, else None.
"""
# Start by assuming failure:
page_messages = None
try:
# Standard token-refresh check:
await self.__ensure_token()
# Build the needed params:
params_json = {
"maxResults": 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.",
self.__user_email,
count,
next_page_token
)
api_response = await self.__http_client.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/messages",
headers = {"Authorization": f"Bearer {self.__credentials.token}"},
params = params_json
)
# If the API call failed:
if api_response.status_code not in [200]: return page_messages
# Else we format the response:
api_json = api_response.json()
page_messages = {
"messages": {m.pop("id"): m for m in api_json.get("messages", [])},
"nextPageToken": api_json.get("nextPageToken"),
"resultSizeEstimate": api_json["resultSizeEstimate"],
}
# In case something goes wrong along the way:
except Exception as exception:
if raise_exception: raise
self._printer(exception)
page_messages = None
# Done here:
return page_messages
async def list_messages(
self,
count: int = 100,
query: str = None,
label_ids: List[str] | str = None,
include_spam_and_trash: bool = False,
raise_exception: bool = False
) -> Dict[str, Any]:
"""
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 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 raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be
suppressed.
:return: The messages that matched the given conditions if the call was successful, else None.
"""
# Start by assuming failure:
messages = None
try:
# Standard token-refresh check:
await self.__ensure_token()
# We convert the messages to a dict:
messages = {
"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(count / max_per_call))
last_iteration_count = count - int((max_per_call * (iterations_needed - 1)))
# Run the loop those many times:
next_page_token = None
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 = count
# Retrieve the messages for this page:
iteration_messages = await self.__list_messages_on_page(
count = iteration_count,
query = query,
label_ids = label_ids,
include_spam_and_trash = include_spam_and_trash,
next_page_token = next_page_token,
raise_exception = True
)
# Check if no data was received:
if iteration_messages is None: break
if not iteration_messages.get("messages"): break
# Now that we know that messages were received:
for k, v in iteration_messages["messages"].items(): messages["messages"][k] = v
results_size_estimate += iteration_messages["resultSizeEstimate"]
# If there is no next page after this, we break out of the loop:
next_page_token = iteration_messages["nextPageToken"]
if next_page_token is None: break
# Format the final response:
messages["nextPageToken"] = next_page_token
messages["resultSizeEstimate"] = results_size_estimate
# In case something goes wrong along the way:
except Exception as exception:
if raise_exception: raise
self._printer(exception)
messages = None
# Done here:
return messages
async def get_message(
self,
message_id: str,
return_raw = False,
raise_exception: bool = False
) -> Dict[str, Any] | str | None:
"""
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 message_id: The id that Google assigned to the message.
:param return_raw: Whether you want the raw message or the formatted message.
:param raise_exception: If set to True, any exceptions that occur will be propagated, else they wil be
suppressed.
: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:
message = None
try:
# Standard token-refresh check:
await self.__ensure_token()
# Make the API call:
if not self._debug_only_errors: self._printer("Getting One Message.", self.__user_email)
api_response = await self.__http_client.get(
url = f"https://gmail.googleapis.com/gmail/v1/users/{self.__user_email}/messages/{message_id}",
headers = {"Authorization": f"Bearer {self.__credentials.token}"},
params = {"format": "raw"}
)
# If the API call failed:
if api_response.status_code not in [200]: return message
# Extract the raw message and respond based on the request:
api_json = api_response.json()
message = base64.urlsafe_b64decode(api_json["raw"])
if return_raw: message = message.decode()
else:
message = mail_parser.parse(message)
message["labels"] = api_json["labelIds"]
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:
except Exception as exception:
if raise_exception: raise
self._printer(exception)
message = None
# Done here:
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
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
import dateparser
# 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)
my_oauth = GoogleOAuth(
config = secrets_dict,
redirect_url = r"https://api.thecaoffice.com/converse/mail/callback/gmail",
debug = True,
debug_prefix = "OAuth (Goog) | ",
)
tokens = {
"access_token": "ya29.a0AeDClZAYoo85BXRId_n-hwo_amKshzi46c33GaJcsZZvGB7A7OGU2RFYcWBM_BleNBfAFUSJP2NHAvmd7Nsp_U5Kg68hXSy0iO99PNTm3pvKrJSzbkA-rXsVLsCnBIfPUMyNt2nOOVJmGwm17DNN0jAELkm1fPNTju7SZzmuaCgYKAZwSARMSFQHGX2Mis8TZui2rZT1gKySVds-N0w0175",
"refresh_token": "1//0gnqzjMf9YT19CgYIARAAGBASNgF-L9Ir3rcY37nGrV45XyOUBRllEH7Txui7T1JbwevlmDoNw7PuMu149cCWQSwsScuKaZusUQ",
"expires_in": 3539,
"expires_at": 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(
http_client = test_client,
credentials = await my_oauth.credentials_from_tokens(tokens = tokens),
debug_only_errors = False
)
await my_gmail.initialize()
print(await my_gmail.user_email)
print(await my_gmail.user_profile)
# print(json.to_string(await my_gmail.list_labels()))
# print(json.to_string(await my_gmail.get_label(label_id = "Label_3")))
# print(json.to_string(await my_gmail.update_label(
# label_id = "Label_3",
# label_name = "Updated Label 123",
# label_background_color = "#cc3a21"
# )))
#
# print(json.to_string(await my_gmail.get_label(label_id = "Label_3")))
# print(json.to_string(await my_gmail.create_label(
# label_name = "Bye - Bye !",
# label_text_color = "#fbc8d9",
# label_background_color = "#7a2e0b"
# )))
# 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.get_message(
# message_id = "19367930033154ca",
# return_raw = False
# ),
# 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())
utils_v2/goog/gmail/gmail_client.py
+959
View File
@@ -0,0 +1,959 @@
"""
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
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 ***
# ***** ****
# *****************************************************************************************************************
# --- 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:
"""
To get the list of labels of this user.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users/getProfile
: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("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 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.pop("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,
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 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": 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, 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.pop("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,
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 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(count / max_per_call))
last_iteration_count = 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 = count
# Retrieve the messages for this page:
iteration_response = await self.__list_messages_on_page(
tokens = tokens,
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"])
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)
# Read the user's tokens that will give you app access to that user's account:
tokens_file = r"../../../creds/goog/user/test_user_tokens_20241125.json"
tokens_dict = json.from_file(tokens_file)
test_tokens = GoogleAuthTokens(**tokens_dict)
async def main():
# Create an instance of the client:
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
)
# Create a sample mail:
my_mail = GMailMessage(
from_email = "pskhushal@gmail.com",
to_email = "orangebhopli@gmail.com",
subject = "Re: Bhopli is the best! (Thread Test)",
cc_emails = None,
bcc_emails = None
)
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"../../../data/images/cat_petting.png")
my_mail.add_attachment(r"../../../data/pdf/sample_label.pdf")
# print(my_mail.get_raw_message(as_base64 = False))
# Test some feature:
response = await my_gmail.send_message(
tokens = test_tokens,
message = my_mail,
thread_id = "1936caffb67996d3"
)
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())
utils_v2/goog/gmail/gmail_message.py
+322
View File
@@ -0,0 +1,322 @@
"""
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:
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
# For random strings:
import string
import random
# For system-level activities:
import os
# For working with files in RAM:
import io
# To work with Base64 encoding:
import base64
# To work with datatypes:
from typing import List
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class GMailMessage:
def __init__(
self,
from_email: str,
to_email: str,
subject: str,
cc_emails: List[str] = None,
bcc_emails: List[str] = None
):
"""
Create an instance of the message that you would like to send.
:param from_email: The EMail ID of the sender.
:param to_email: The EMail ID of the recipient.
:param subject: The subject of the mail.
:param cc_emails: A list of recipients to add to the CC section.
:param bcc_emails: A list of recipients to add to the BCC section.
"""
# Create the instance of the message:
self.message = MIMEMultipart()
self.message["From"] = from_email
self.message["To"] = to_email
self.message["Subject"] = subject
if cc_emails: self.message["CC"] = ",".join(cc_emails)
if bcc_emails: self.message["BCC"] = ",".join(bcc_emails)
# Note down the values for accessing later:
self.__from = from_email
self.__to = to_email
self.__cc = cc_emails,
self.__bcc = bcc_emails
self.__subject = subject
# ┏┓ •
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
# ┛
@property
def from_mail(self):
return self.__from
@property
def to_mail(self):
return self.__to
@property
def cc_mails(self):
return self.__cc
@property
def bcc_mails(self):
return self.__bcc
@property
def subject(self):
return self.__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: str | io.BytesIO,
file_name: str = None
):
"""
Add a file as an attachment to the mail. This file, even if possible, will not be rendered on the screen in-line
with the body. It will be made available as a download.
:param attachment_file: The file that you would like to attach.
:param file_name: The name of the file. This is the same name by which it will be downloaded. You need not
specify this if the input file is specified as a path. Needed when you give the input file as a buffer.
:return:
"""
# 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 isinstance(attachment_file, 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:
elif isinstance(attachment_file, 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_raw_message(
self,
as_base64: bool = True
):
"""
Get the raw string dump from the current contents of the message. This text will be compliant with RFC 5322 and
RFC 2045 (among others).
:param as_base64: If set to True, the response will be a URL-safe B64 output, else it'll be a raw string.
:return: The standardized raw text dump. either as a raw string or as a Base64 (url-safe) string.
"""
if not as_base64: return self.message
else: return base64.urlsafe_b64encode(self.message.as_bytes()).decode()
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
my_mail = GMailMessage(
from_email = "sender@gmail.com",
to_email = "recipient@gmail.com",
subject = "Bhopli is the best!",
cc_emails = None,
bcc_emails = None
)
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"../../../data/images/cat_petting.png")
my_mail.add_attachment(r"../../../data/pdf/sample_label.pdf")
print(my_mail.get_raw_message(as_base64 = True))
utils_v2/goog/gmail/gmail_v2.py
-464
View File
@@ -1,464 +0,0 @@
"""
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:
"""
To get the list of labels of this user.
DOCUMENTATION:
1. https://developers.google.com/gmail/api/reference/rest/v1/users/getProfile
: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("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 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.pop("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("Getting 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("Getting 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
# *****************************************************************************************************************
# ***** ****
# *** 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/data/api_call.py
+1
View File
@@ -107,6 +107,7 @@ class GoogleApiResponse(BaseModel):
message += f"*URL:*\n`{self.url}`\n\n" message += f"*URL:*\n`{self.url}`\n\n"
message += f"*METHOD:*\n`{self.method}`\n\n" message += f"*METHOD:*\n`{self.method}`\n\n"
message += f"*RESPONSE:*\n`{self.response}`\n\n" message += f"*RESPONSE:*\n`{self.response}`\n\n"
message += f"*MESSAGE:*\n`{self.message}`\n\n"
message += f"*EXCEPTION:*\n`{self.exception.__class__.__name__}: {str(self.exception)}`\n\n" message += f"*EXCEPTION:*\n`{self.exception.__class__.__name__}: {str(self.exception)}`\n\n"
return message return message
utils_v2/oauth/services/goog.py
+1 -1
View File
@@ -363,7 +363,7 @@ if __name__ == "__main__":
import dateparser import dateparser
secrets_file = r"../../../creds/google_tcaoff_test_oauth_20241125.json" secrets_file = r"../../../creds/goog/app/google_tcaoff_test_oauth_20241125.json"
secrets_dict = json.from_file(secrets_file) secrets_dict = json.from_file(secrets_file)
my_goog = GoogleOAuth( my_goog = GoogleOAuth(