I have built this Django Package which is an alternative for sending mail to users using Gmail API. Everything is working fine. But the verify(self, request) method inside gmailer/gmail.py
is not feeling great with the way I coded.
I need to optimize this code and understand if there is a better style of coding that I can adapt here regarding error handling and writing testcases for this code. Here is the source to the Django package.
from base64 import urlsafe_b64encode
from django.conf import settings
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from googleapiclient.errors import HttpError
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
_GMAIL_CONSTS = {
'GMAIL_SECRET': 'google_client_secret.json',
'GMAIL_SCOPES': [
"https://www.googleapis.com/auth/gmail.metadata",
"https://www.googleapis.com/auth/gmail.send",
],
'GMAIL_REDIRECT': 'http://localhost:8000/gmailer/verify',
'GMAIL_USER': 'Django Mail Admin'
}
class Gmail:
def __init__(self, client_secrets_file, scopes, redirect_uri, user):
self.activated = False
self.user = user
self.outbox = []
self.flow = InstalledAppFlow.from_client_secrets_file(
client_secrets_file=client_secrets_file,
scopes=scopes,
redirect_uri=redirect_uri)
def authorize(self):
return self.flow.authorization_url()
def verify(self, request):
code = request.GET.get('code', '')
state = request.GET.get('state', '')
if code and 'oauth_state' in request.session and state == request.session['oauth_state']:
try:
self.flow.fetch_token(code=code)
self.credentials = self.flow.credentials
self.service = build(
'gmail', 'v1', credentials=self.credentials)
self.email = (self.service.users().getProfile(
userId="me").execute())['emailAddress']
self.activated = True
return {
'user': self.user,
'email': self.email,
'credentials': {
'token': self.credentials.token,
'refresh_token': self.credentials.refresh_token,
'token_uri': self.credentials.token_uri,
'client_id': self.credentials.client_id,
'client_secret': self.credentials.client_secret,
'scopes': self.credentials.scopes,
},
}
except Exception as err:
self.revoke()
return {'message': 'Unable to authorize request', 'error': str(err)}
else:
raise self.StateError()
def revoke(self):
if hasattr(self, 'service'):
self.service.close()
self.activated = False
self.credentials = None
self.service = None
self.email = ''
self.outbox = []
def _create_message(self, subject, message_text, from_email, recipient_list, html):
message = MIMEMultipart()
message['to'] = ', '.join(recipient_list)
message['from'] = from_email
message['subject'] = subject
if html:
message.attach(MIMEText(message_text, 'html'))
else:
message.attach(MIMEText(message_text, 'plain'))
raw = urlsafe_b64encode(message.as_bytes()).decode()
return {'raw': raw}
def send_mail(self, subject, message, recipient_list, html=False):
if self.activated:
try:
body = self._create_message(
subject,
message,
self.user + " <" + self.email + ">",
recipient_list,
html)
sent_message = (self.service.users().messages().send(
userId="me",
body=body).execute())
print('Message sent with id: %s' % sent_message['id'])
return sent_message
except HttpError as error:
pass
else:
raise self.UnauthorizedAPIError()
def test_mail(self, subject=None, message=None, recipient_list=None):
if self.activated:
try:
sub = subject or "Django Google Mailer"
msg = message or "Hi,\n\nWelcome to Django Site"
to = recipient_list or [self.email]
sent_message = self.send_mail(sub, msg, to)
self.outbox.append(sent_message)
except Exception as e:
print(e)
else:
raise self.UnauthorizedAPIError()
@staticmethod
def add_settings():
for attr in ['GMAIL_SECRET', 'GMAIL_SCOPES', 'GMAIL_REDIRECT', 'GMAIL_USER']:
if not hasattr(settings, attr):
setattr(settings, attr, _GMAIL_CONSTS[attr])
installed_apps = getattr(settings, 'INSTALLED_APPS', [])
installed_apps += ['gmailer', ]
setattr(settings, 'INSTALLED_APPS', installed_apps)
return True
class SettingError(Exception):
def __init__(self, setting_name, error=None):
self.message = setting_name + ' is missing/misconfigured inside Django Settings'
self.error = error
super().__init__(self.message)
class StateError(Exception):
def __init__(self, message=None, error=None):
self.message = message or "The state/code is not valid. Check the verification url."
self.error = error
super().__init__(self.message)
class UnauthorizedAPIError(Exception):
def __init__(self, message=None, error=None):
self.message = message or "Gmail API Service is not authorized. Contact site administrator."
self.error = error
super().__init__(self.message)
if Gmail.add_settings():
mailer = Gmail(
client_secrets_file=settings.GMAIL_SECRET,
scopes=settings.GMAIL_SCOPES,
redirect_uri=settings.GMAIL_REDIRECT,
user=settings.GMAIL_USER)