- [How to Use Gmail API in Python](https://www.thepythoncode.com/article/use-gmail-api-in-python). ([code](general/gmail-api))

To use the scripts here:

- Create your credentials file in Google API dashboard and putting it in the current directory, follow [this tutorial](https://www.thepythoncode.com/article/use-gmail-api-in-python) for detailed information.

- `pip3 install -r requirements.txt`

- Change `our_email` variable in `common.py` to your gmail address.

- To send emails, use the `send_emails.py` script:

```

import os

import pickle

from googleapiclient.discovery import build

from google_auth_oauthlib.flow import InstalledAppFlow

from google.auth.transport.requests import Request

SCOPES = ['https://mail.google.com/']

our_email = 'our_email@gmail.com'

def gmail_authenticate():

creds = None

# the file token.pickle stores the user's access and refresh tokens, and is

# created automatically when the authorization flow completes for the first time

if os.path.exists("token.pickle"):

with open("token.pickle", "rb") as token:

creds = pickle.load(token)

# if there are no (valid) credentials availablle, let the user log in.

if not creds or not creds.valid:

if creds and creds.expired and creds.refresh_token:

creds.refresh(Request())

else:

flow = InstalledAppFlow.from_client_secrets_file('credentials.json', SCOPES)

creds = flow.run_local_server(port=0)

# save the credentials for the next run

with open("token.pickle", "wb") as token:

pickle.dump(creds, token)

return build('gmail', 'v1', credentials=creds)

def search_messages(service, query):

result = service.users().messages().list(userId='me',q=query).execute()

messages = [ ]

if 'messages' in result:

messages.extend(result['messages'])

while 'nextPageToken' in result:

page_token = result['nextPageToken']

result = service.users().messages().list(userId='me',q=query, pageToken=page_token).execute()

if 'messages' in result:

messages.extend(result['messages'])

return messages

from common import gmail_authenticate, search_messages

def delete_messages(service, query):

messages_to_delete = search_messages(service, query)

# it's possible to delete a single message with the delete API, like this:

# service.users().messages().delete(userId='me', id=msg['id'])

# but it's also possible to delete all the selected messages with one query, batchDelete

return service.users().messages().batchDelete(

userId='me',

body={

'ids': [ msg['id'] for msg in messages_to_delete]

}

).execute()

if __name__ == "__main__":

import sys

service = gmail_authenticate()

delete_messages(service, sys.argv[1])

import argparse

from common import search_messages, gmail_authenticate

from read_emails import read_message

from send_emails import send_message

from delete_emails import delete_messages

from mark_emails import mark_as_read, mark_as_unread

if __name__ == '__main__':

parser = argparse.ArgumentParser(description="Send/Search/Delete/Mark messages using gmail's API.")

subparsers = parser.add_subparsers(help='Subcommands')

parser_1 = subparsers.add_parser('send', help='Send an email')

parser_1.add_argument('destination', type=str, help='The destination email address')

parser_1.add_argument('subject', type=str, help='The subject of the email')

parser_1.add_argument('body', type=str, help='The body of the email')

parser_1.add_argument('files', type=str, help='email attachments', nargs='+')

parser_1.set_defaults(action='send')

parser_2 = subparsers.add_parser('delete', help='Delete a set of emails')

parser_2.add_argument('query', type=str, help='a search query that selects emails to delete')

parser_2.set_defaults(action='delete')

parser_3 = subparsers.add_parser('mark', help='Marks a set of emails as read or unread')

parser_3.add_argument('query', type=str, help='a search query that selects emails to mark')

parser_3.add_argument('read_status', type=bool, help='Whether to mark the message as unread, or as read')

parser_3.set_defaults(action='mark')

parser_4 = subparsers.add_parser('search', help='Marks a set of emails as read or unread')

parser_4.add_argument('query', type=str, help='a search query, which messages to display')

parser_4.set_defaults(action='search')

args = parser.parse_args()

service = gmail_authenticate()

if args.action == 'send':

# TODO: add attachements

send_message(service, args.destination, args.subject, args.body, args.files)

elif args.action == 'delete':

delete_messages(service, args.query)

elif args.action == 'mark':

print(args.unread_status)

if args.read_status:

mark_as_read(service, args.query)

else:

mark_as_unread(service, args.query)

elif args.action == 'search':

results = search_messages(service, args.query)

for msg in results:

read_message(service, msg)

from common import gmail_authenticate, search_messages

def mark_as_read(service, query):

messages_to_mark = search_messages(service, query)

return service.users().messages().batchModify(

userId='me',

body={

'ids': [ msg['id'] for msg in messages_to_mark ],

'removeLabelIds': ['UNREAD']

}

).execute()

def mark_as_unread(service, query):

messages_to_mark = search_messages(service, query)

return service.users().messages().batchModify(

userId='me',

body={

'ids': [ msg['id'] for msg in messages_to_mark ],

'addLabelIds': ['UNREAD']

}

).execute()

if __name__ == "__main__":

import argparse

parser = argparse.ArgumentParser(description="Marks a set of emails as read or unread")

parser.add_argument('query', help='a search query that selects emails to mark')

parser.add_argument("-r", "--read", action="store_true", help='Whether to mark the message as read')

parser.add_argument("-u", "--unread", action="store_true", help='Whether to mark the message as unread')

args = parser.parse_args()

service = gmail_authenticate()

if args.read:

mark_as_read(service, args.query)

elif args.unread:

mark_as_unread(service, args.query)

import os

import sys

from base64 import urlsafe_b64decode

from common import gmail_authenticate, search_messages

def get_size_format(b, factor=1024, suffix="B"):

"""

for unit in ["", "K", "M", "G", "T", "P", "E", "Z"]:

if b < factor:

return f"{b:.2f}{unit}{suffix}"

b /= factor

return f"{b:.2f}Y{suffix}"

def clean(text):

# clean text for creating a folder

return "".join(c if c.isalnum() else "_" for c in text)

def parse_parts(service, parts, folder_name):

"""

if parts:

for part in parts:

filename = part.get("filename")

mimeType = part.get("mimeType")

body = part.get("body")

data = body.get("data")

file_size = body.get("size")

part_headers = part.get("headers")

if part.get("parts"):

# recursively call this function when we see that a part

# has parts inside

parse_parts(service, part.get("parts"), folder_name)

if mimeType == "text/plain":

# if the email part is text plain

if data:

text = urlsafe_b64decode(data).decode()

print(text)

elif mimeType == "text/html":

# if the email part is an HTML content

# save the HTML file and optionally open it in the browser

if not filename:

filename = "index.html"

filepath = os.path.join(folder_name, filename)

print("Saving HTML to", filepath)

with open(filepath, "wb") as f:

f.write(urlsafe_b64decode(data))

else:

# attachment other than a plain text or HTML

for part_header in part_headers:

part_header_name = part_header.get("name")

part_header_value = part_header.get("value")

if part_header_name == "Content-Disposition":

if "attachment" in part_header_value:

# we get the attachment ID

# and make another request to get the attachment itself

print("Saving the file:", filename, "size:", get_size_format(file_size))

attachment_id = body.get("attachmentId")

attachment = service.users().messages() \

.attachments().get(id=attachment_id, userId='me', messageId=msg['id']).execute()

data = attachment.get("data")

filepath = os.path.join(folder_name, filename)

if data:

with open(filepath, "wb") as f:

f.write(urlsafe_b64decode(data))

def read_message(service, message_id):

"""

msg = service.users().messages().get(userId='me', id=message_id['id'], format='full').execute()

# parts can be the message body, or attachments

payload = msg['payload']

headers = payload.get("headers")

parts = payload.get("parts")

folder_name = "email"

if headers:

# this section prints email basic info & creates a folder for the email

for header in headers:

name = header.get("name")

value = header.get("value")

if name == 'From':

# we print the From address

print("From:", value)

if name == "To":

# we print the To address

print("To:", value)

if name == "Subject":

# make a directory with the name of the subject

folder_name = clean(value)

# we will also handle emails with the same subject name

folder_counter = 0

while os.path.isdir(folder_name):

folder_counter += 1

# we have the same folder name, add a number next to it

if folder_name[-1].isdigit() and folder_name[-2] == "_":

folder_name = f"{folder_name[:-2]}_{folder_counter}"

elif folder_name[-2:].isdigit() and folder_name[-3] == "_":

folder_name = f"{folder_name[:-3]}_{folder_counter}"

else:

folder_name = f"{folder_name}_{folder_counter}"

os.mkdir(folder_name)

print("Subject:", value)

if name == "Date":

# we print the date when the message was sent

print("Date:", value)

parse_parts(service, parts, folder_name)

print("="*50)

if __name__ == "__main__":

service = gmail_authenticate()

# get emails that match the query you specify from the command lines

results = search_messages(service, sys.argv[1])

# for each email matched, read it (output plain/text to console & save HTML and attachments)

for msg in results:

read_message(service, msg)

google-api-python-client

google-auth-httplib2