Small Python script using Google and Slack APIs

Question

I've recently been working on a small Python script that I was tasked to do at my new job. It will basically handle the rotation and notification of the on call engineer by populating a Google Calendar with the schedule as well as pinging a Slack channel I created. The script will be run weekly as a cron job and will pull the list from a Google Sheet. This is my first time using both Google and Slack's APIs, so please let me know of any ways in which I could improve my code. Thanks!

# pylint: disable=E1101
"""
Tracks and rotates the on call engineer at REDACTED.
Uses Slack and Google's API to obtain access to the services.
"""
from os import mkdir, path, environ
from os.path import join, dirname
from datetime import datetime, timedelta
import argparse
from httplib2 import Http
from apiclient import discovery
from oauth2client.file import Storage
from oauth2client import client
from oauth2client import tools
from dotenv import load_dotenv
from slackclient import SlackClient
# try getting flags (if any, not sure what this does)
FLAGS = None
# scopes and google stuff
SCOPES = 'https://www.googleapis.com/auth/calendar https://www.googleapis.com/auth/spreadsheets'
CLIENT_SECRET_FILE = 'client_secrets.json'
APPLICATION_NAME = 'On Call Tracker'
SHEET_ID = 'REDACTED'
CALENDAR_ID = 'REDACTED'
# load variables from .env file
DOTENV_PATH = join(dirname(__file__), '.env')
load_dotenv(DOTENV_PATH)
# set slack authentication constants
SLACK_TOKEN = environ.get("SLACK_TOKEN")
SLACK_CLIENT = SlackClient(SLACK_TOKEN)
def get_credentials():
 """Gets valid user credentials from storage.
 If nothing has been stored, or if the stored credentials are invalid,
 the OAuth2 flow is completed to obtain the new credentials.
 Returns:
 Credentials, the obtained credential.
 """
 home_dir = path.expanduser('~')
 credential_dir = path.join(home_dir, '.credentials')
 if not path.exists(credential_dir):
 mkdir(credential_dir)
 credential_path = path.join(credential_dir,
 'calendar-python-quickstart.json')
 store = Storage(credential_path)
 credentials = store.get()
 if not credentials or credentials.invalid:
 flow = client.flow_from_clientsecrets(CLIENT_SECRET_FILE, SCOPES)
 flow.user_agent = APPLICATION_NAME
 if FLAGS:
 credentials = tools.run_flow(flow, store, FLAGS)
 print 'Storing credentials to ' + credential_path
 return credentials
def add_event(event_name, service, start_date):
 """Adds an event to the google calendar with the given name.
 Args:
 event_name (str): the name of the calendar event
 service (obj): the google calendar api service
 """
 end_date = start_date + timedelta((2 - start_date.weekday()) % 7 + 1)
 description = """If unreachable, please contact either:
 foo bar: [email protected] [phone]
 bar foo: [email protected] [phone]"""
 event = {
 'summary': event_name + ' On Call',
 'start': {
 'date': datetime.strftime(start_date, "%Y-%m-%d")
 },
 'end': {
 'date': datetime.strftime(end_date, "%Y-%m-%d")
 },
 'description': description
 }
 event = service.events().insert(calendarId=CALENDAR_ID, body=event).execute()
def del_all_events(service):
 """Deletes all on-call events from the current time onwards.
 Args:
 service (obj): the Google Calendar API service
 """
 cur_time = datetime.utcnow().isoformat("T") + "Z"
 events_list = service.events().list(calendarId=CALENDAR_ID,
 timeMin=cur_time, q='On Call').execute()['items']
 on_call_events = [event for event in events_list if event['creator']['email'] == '[email protected]']
 for event in on_call_events:
 event_id = event['id']
 service.events().delete(calendarId=CALENDAR_ID, eventId=event_id).execute()
 print "All events successfully deleted."
def rotate_names_in_sheet(value_range, service):
 """Rotates a given column of cells in the Google Spreadsheet using Google's API.
 Args:
 value_range (obj): a ValueRange object from the Google Spreadsheet
 service (obj): the Google API service object
 Returns:
 The list of names, after having been rotated.
 """
 values = value_range.get('values', [])
 # rotate the list
 values = values[1:] + values[:1]
 # create new ValueRange instance
 request_data = {
 "values": values
 }
 service.spreadsheets().values().update(
 spreadsheetId=SHEET_ID, range='A2:C', valueInputOption='USER_ENTERED', body=request_data
 ).execute()
 return [value[0] for value in values]
def send_message(channel_id, message_text, user):
 """Sends the given message to the given slack channel.
 Args:
 channel_id (str): the ID of the channel to post a message in
 message_text (str): the message text to send_message
 user (bool): is the message sender a user or not
 The user argument determines if the message will be sent as a bot or as the
 currently authorized user.
 """
 if not user:
 SLACK_CLIENT.api_call(
 "chat.postMessage",
 channel=channel_id,
 text=message_text,
 username='Not_A_Robot',
 as_user=user,
 icon_emoji=':robot_face:'
 )
 else:
 SLACK_CLIENT.api_call(
 "chat.postMessage",
 channel=channel_id,
 text=message_text,
 as_user=user
 )
def list_channels():
 """Lists all the private message channels of the authorized user"""
 channels_call = SLACK_CLIENT.api_call("groups.list")
 if channels_call['ok']:
 return channels_call['groups']
 return None
def ping_slack(on_call_name, chan_list):
 """Pings a Slack channel with to alert the channel with the new on call engineer.
 Args:
 on_call_name (str): the name of the new on call engineer
 chan_list (str): the list of channels as a json object
 """
 names = []
 channel_ids = []
 for chan in chan_list:
 names.append(chan['name'])
 channel_ids.append(chan['id'])
 channels_dict = dict(zip(names, channel_ids))
 send_message(channels_dict['on_call_engineers'],
 on_call_name + " is on call for this week.", False)
def main():
 """Rotates names of on call engineers in a Google Spreadsheet and updates/notifies the team.
 Uses Google's calendar and sheets API as well as Slack's API to alert a Slack channel.
 """
 # google shizz
 creds = get_credentials()
 http = creds.authorize(Http())
 calendar = discovery.build('calendar', 'v3', http=http)
 sheets = discovery.build('sheets', 'v4', http=http)
 value_range = sheets.spreadsheets().values().get(
 spreadsheetId=SHEET_ID, range='A2:C'
 ).execute()
 parser = argparse.ArgumentParser()
 group = parser.add_mutually_exclusive_group(required=True)
 group.add_argument('-u', '--update', help='rotate and update/notify', action='store_true')
 group.add_argument('-c', '--clear',
 help='clears all on-call events from future', action='store_true')
 args = parser.parse_args()
 if args.clear or args.update:
 del_all_events(calendar)
 if args.update:
 list_of_names = rotate_names_in_sheet(value_range, sheets)
 print "List rotated, current new on call is: " + list_of_names[0]
 start_date = datetime.today()
 # display events for next cycle
 for name in list_of_names:
 add_event(name, calendar, start_date)
 start_date = start_date + timedelta((2 - start_date.weekday()) % 7 + 1)
 # slack shizz
 channels = list_channels()
 ping_slack(list_of_names[0], channels)
if __name__ == "__main__":
 main()

Answer 1

• Compability with Python 3 is a good idea, in particular the print statement should be a function call instead.
• path.join can also be directly done on the path to save a line.
• If variables are only used once, maybe just inline them (e.g. DOTENV_PATH).
• Some of the file names and other strings are hardcoded, some are in constants at the top - why the distinction? Worse, e.g. 'A2:C' isn't a constant but was written out twice!
• return None at the end of a function is the same as not having a return statement.

• In ping_slack first splitting, then zipping the values is more work than just having a single dictionary comprehension, e.g.:

  channels_dict = {chan['name']: chan['id'] for chan in chan_list}
  

  However given that only a single entry is used anyway this does more work than just saving that single value, but looks perhaps more elegant.

  Lastly the docstring is wrong, the chan_list argument is a list, not a string.

More general things:

• Some of the function names, like del_all_events are shortened for no real reason, whereas rotate_names_in_sheet is long and descriptive. I'd choose either style and stick with it.
• In send_message, the user argument should be a bit more descriptive, perhaps as_user to signify that it's basically a flag, not a "user" object. Oh right, that's also that the Slack API uses, so why not adhere to that.
• The error checking and the way that different functionality is split up between components could be improved. E.g. rotate_names_in_sheet uses a default value ([]) for the values value, but in the main function there'll be an index out of range exception if that case happens at any time - it'd be better if that situation is handled earlier, at the point that it occurs, i.e. if value is missing an exception should be raised (or alternatively the main function fixed such that it can cope with the missing name).
• Similarly a couple of expressions, e.g. service.events, come up more than once, perhaps some refactoring such that the code around that is more succinct would be good.
• The date calculations are somewhat opaque. Using a better API would make it probably more readable, but I don't have a suggestion as to which library that could be done with.

• AJwr: Thanks for the help, these were a lot of great suggestions that I will definitely be implementing. About the constants at the top of the file, those are loaded in from a .env file as some contain sensitive information that I would not want to hardcode, even if it was only used once. Thanks again!