2
\$\begingroup\$

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)
\$\endgroup\$

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Browse other questions tagged or ask your own question.