Make a script to call pager duty's API and update on call

Question

I made this script it works, I'm just looking for ways to make it better/reading to get better.

#!/usr/bin/python3
import requests, logging, json, subprocess, re, smtplib
from datetime import datetime, date
# logger things
log_file = '/var/log/pagerduty.log'
## logging things
logging.basicConfig(level=logging.INFO, format='%(process)d-%(asctime)s-%(levelname)s-%(funcName)s %(message)s',
 filename=log_file)
###
today = datetime.now()
api_key = '<removed>'
file_path = '<removed>'
# teams_webhook_url = <removed>'
headers = {
 "Accept": "application/vnd.pagerduty+json;version=2",
 "Authorization": f"Token token={api_key}",
 "Content-Type": "application/json"
}
userInfo = {
 "Override": {
 "id": None,
 "name": None,
 "date_start": None,
 "date_end": None,
 "mobile": None,
 "turned_back": None
 },
 "onCall": {
 "id": None,
 "name": None,
 "date": None,
 "mobile": None,
 }
}
schedule_id = '<removed>'
start_time = today.replace(hour=0, minute=0)
end_time = today.replace(hour=23, minute=59)
print(start_time)
with open(file_path, 'r') as file:
 oldFacts = json.load(file)
params = {
 'since': start_time.strftime('%Y-%m-%dT%H:%M:%SZ'),
 'until': end_time.strftime('%Y-%m-%dT%H:%M:%SZ')
}
print(params)
# gets whos on call for real to get there mobile number
def whosOnCall():
 endpointoncall = f'https://api.pagerduty.com/schedules/{schedule_id}'
 responseOnCall = requests.get(endpointoncall, headers=headers, params=params)
 if responseOnCall.status_code == 200:
 oncallrawdata = responseOnCall.json()
 schedule_entries = oncallrawdata['schedule']['schedule_layers'][0]['rendered_schedule_entries']
 for facts in schedule_entries:
 logging.info(facts)
 userInfo['onCall']['id'] = facts['user']['id']
 userInfo['onCall']['name'] = facts['user']['summary']
 break
 else:
 logging.info(f"Failed to fetch overrides. Status code: {responseOnCall.status_code}")
 logging.info(responseOnCall.text)
 message = {"text": "Failed to fetch overrides. Status code: " + str(responseOnCall.status_code) + "🦭🦭 "}
 # teamsMessage(message)
# gett the overrides and userID to get the number later
def userid():
 endpointOverrides = f'https://api.pagerduty.com/schedules/{schedule_id}/overrides'
 response = requests.get(endpointOverrides, headers=headers, params=params)
 if response.status_code == 200:
 overrides_data = response.json()
 overrides = overrides_data['overrides']
 if not overrides:
 logging.info(f"no call forwarded needed")
 return False
 else:
 # returns a list of dics for overrides
 for override in overrides:
 save_override_info(override)
 return True
 else:
 logging.info(f"Failed to fetch overrides. Status code: {response.status_code}")
 logging.info(response.text)
 message = {"text": "Failed to fetch overrides. Status code: " + str({response.status_code}) + " 🦭🦭 "}
 # teamsMessage(message)
def save_override_info(override):
 date_start = str(override['start'])
 date_end = str(override['end'])
 userInfo['Override']['id'] = override['user']['id']
 userInfo['Override']['date_start'] = date_start[:16]
 userInfo['Override']['date_end'] = date_end[:16]
 userInfo['Override']['name'] = override['user']['summary']
 logging.info(f" this is the info override found {userInfo}")
 return True
# puts the number from pager duty
def phonenumber(userID, whatType):
 endpointMobileNumber = f"https://api.pagerduty.com/users/{userID}/contact_methods/"
 responseMobile = requests.get(endpointMobileNumber, headers=headers)
 if responseMobile.status_code == 200:
 mobileData = responseMobile.json()
 for contactMethod in mobileData['contact_methods']:
 if contactMethod['type'] == 'phone_contact_method':
 userInfo[whatType]['mobile'] = checknumber(contactMethod['address'])
 elif contactMethod['type'] == 'email_contact_method':
 userInfo[whatType]['email'] = contactMethod['address']
 else:
 logging.info(f"Failed to fetch overrides. Status code: {responseMobile.status_code}")
 message = {"text": "Failed to fetch overrides. Status code: " + str(responseMobile.status_code) + " 🦭🦭"}
 # teamsMessage(message)
 logging.info(responseMobile.text)
## email soooooon
def emails(whattype, to, number, overname, output):
 sender = '<removed>'
 try:
 if whattype in ['Override', 'onCall_return']:
 message_subject = 'Whoah man, you\'re on call'
 message_body = {
 f'🦭🦭! \n\n Call forward for on call number <removed> has been set to your mobile {number} '
 f'{output}'}
 else:
 message_subject = "Whoah man, override found you "
 message_body = {f'You are not on call 🦭🦭! \n\n Call forward for on call number <removed> has been '
 f'set to {overname} person {output}'}
 message = f'From: On Call<{sender}>\nTo: To Person {to}\nSubject: {message_subject}\n\n{message_body}'
 message_utf = message.encode('utf-8')
 session = smtplib.SMTP(<removed>')
 session.sendmail(sender, '<removed>'', message_utf)
 logging.info(f'Successfully sent email to {to}')
 except Exception as e:
 logging.info(f"Error: unable to send email {e}")
# def teamsMessage(message):
# response = requests.post(teams_webhook_url, json=message)
# if response.status_code == 200:
# logging.info("Message as been sent")
# else:
# logging.info(f"Failed to send message. status code {response.status_code}")
# forwarding the on call 0800 number
def forwardNumbers(number):
 try:
 cmd = ["/usr/local/bin/updateoncallforward", number]
 run = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
 stdout, stderr = run.communicate()
 stdout_without_password = re.sub(r'"password": ".*?"', '"password": "********"', stdout)
 logging.info(f"Subprocess Output: {stdout_without_password}")
 return stdout_without_password
 except FileNotFoundError:
 message = {"text": "updateoncallforward executable not found. help me SOS 🦭🦭"}
 # teamsMessage(message)
 logging.info("updateoncallforward executable not found. 🦭🦭")
 except subprocess.CalledProcessError as e:
 message = {"text": "python call forward script exited with non-zero exit code: " + str(e.returncode) + "🦭🦭"}
 # teamsMessage(message)
 logging.info(f"updateoncallforward script exited with non-zero exit code: {e.returncode}")
# checks the number from pager duty as the numbers starts without the starting 0 adds that to the number
def checknumber(phone_number):
 pattern = r'^02\d{9}|02\d{8}$'
 if not str(phone_number).startswith('0'):
 phone_number = '0' + phone_number
 if re.match(pattern, phone_number):
 return phone_number
 else:
 logging.info(f"somthing is wrong with the phone number the number is question is {phone_number}")
 message = {"text": "somthing is wrong with the phone number the number is question is {phone_number}"}
 # teamsMessage(message)
def main():
 whosOnCall()
 output = ""
 current_server_time = datetime.now()
 start_time = current_server_time.strftime("%Y-%m-%dT%H:%M")
 print(start_time)
 override_needed = userid()
 if start_time == userInfo['Override']['date_start']:
 if override_needed is True and (oldFacts['Override']['turned_back'] == 0 or oldFacts['Override']['name']
 != userInfo['Override']['name']):
 for dic in userInfo:
 phonenumber(userInfo[dic]['id'], dic)
 output = forwardNumbers(userInfo['Override']['mobile'])
 for emailsAddresses in userInfo:
 if emailsAddresses == "Override":
 emails(emailsAddresses, userInfo[emailsAddresses]['email'], userInfo[emailsAddresses]['mobile'],
 userInfo[emailsAddresses]['name'], output)
 else:
 emails(emailsAddresses, '<removed>', userInfo[emailsAddresses]['mobile'],
 userInfo[emailsAddresses]['name'], output)
 
#emails('onCall', userInfo['onCall']['email'], userInfo['Override']['mobile'],
# userInfo['Override']['name'], output)
 message = {"text": "override found for today forwarded on call number from " + str(userInfo['onCall']['name'])
 + "to " + str(userInfo['Override']['name']) + str(output)}
 #teamsMessage(message)
 userInfo['Override']['turned_back'] = 1
 with open(file_path, 'w') as file_override:
 json.dump(userInfo, file_override, indent=4)
 elif (
 override_needed is False
 and oldFacts['Override']['turned_back'] == 1
 and start_time == oldFacts['Override']['date_end']
 ):
 output = forwardNumbers(oldFacts['onCall']['mobile'])
 for emailsAddresses in oldFacts:
 if emailsAddresses == "Override":
 emails(emailsAddresses, oldFacts[emailsAddresses]['email'], oldFacts[emailsAddresses]['mobile'],
 oldFacts[emailsAddresses]['name'], output)
 else:
 emails(emailsAddresses, '<removed>', oldFacts[emailsAddresses]['mobile'],
 oldFacts[emailsAddresses]['name'], output)
 #emails('Override_return', oldFacts['Override']['email'], oldFacts['Override']['mobile'],
 # oldFacts['Override']['name'], output)
 #emails('onCall_return', oldFacts['onCall']['email'], oldFacts['Override']['mobile'],
 # oldFacts['Override']['name'], output)
 message = {"text": "Override returned to " + str(oldFacts['onCall']['name']) + "that was taken over from " +
 str(oldFacts['Override']['name']) + " could we just double check this worked 🦭🦭"}
 # teamsMessage(message)
 oldFacts['Override']['turned_back'] = 0
 with open(file_path, 'w') as file:
 json.dump(oldFacts, file, indent=4)
 logging.info(f"on call returned to {oldFacts['onCall']['name']}")
 else:
 logging.info("nothing to be done")
 logging.info("-" * 60)
if __name__ == "__main__":
 main()
# alert help im lost

Also when it finds an override it runs a different script that updates the on-call number on net and also updates the offset number, and forwards the output of that script so you know it worked.

• Toby Speight: "I made this script it works", but you didn't tell us what "works" means. Do you have a definition of the intended behaviour that we can refer to? Perhaps some unit tests, or a functional specification, or just some sample inputs and corresponding output?
• Alex Orbit: hello oh it pulls whos on call then the overrides say someone is taking over on call on call info

Answer 1

$PATH + shebang

#! /usr/bin/python3

That's kind of a weird shebang. It mandates that we use the system python, rather than one maintained by {conda, poetry, pip venv}. We are importing requests from pypi, so presumably we did a global $ sudo pip install requests. That doesn't scale well when we have many such packages installed, and need to worry about DLL-hell version conflicts. Prefer to maintain per-project environments of such libraries.

Which implies you will have more than one python executable file, and will need to activate a given environment to use it, so the output of $ which python will change. Here is a shebang which supports such usage:

#! /usr/bin/env python

The env program is always in the same spot. And it will use $PATH to locate and execute a python interpreter, possibly not the one in /usr/bin.

---

comments tell the why, not the how

# logger things
...
## logging things

Elide useless comments -- the code already eloquently spoke of what's happening. Save comments for when you have something non-obvious to share with the Gentle Reader.

---

appropriate names

I'm worried about truth-in-advertising here:

today = datetime.now()

I would expect that to be a date-granularity date, rather than a fine-grained datetime.

Pick a new name, perhaps current_time, or simply now.

---

credentials don't go in source code

api_key = '<removed>'

Secrets don't belong checked into git. Prefer to assign api_key = os.environ['PAGERDUTY_API_KEY'], or perhaps read it from a YAML / JSON config file.

---

day versus time

userInfo = {
 "Override": {
 ...
 "date_start": ...
 "date_end": ...
 ...
 "onCall": {
 ...
 "date": ...

Similar to above, I am concerned that perhaps you meant "timestamp" where you wrote "date". That is, I would expect these to have H:M:S granularity, yet the identifier suggests they only represent Y-M-D. Names matter. Pick accurate ones.

---

lint

 oldFacts = json.load(file)

Pep-8 asks that you spell it old_facts, please.

Similarly for whos_on_call, response_on_call, etc.

---

timezone offset

 'since': start_time.strftime('%Y-%m-%dT%H:%M:%SZ'),
 'until': end_time.strftime('%Y-%m-%dT%H:%M:%SZ')

Kudos! I love to see that trailing Zulu. Now the programs, the data, the people, are all very clear that the zone offset is zero. Good deal. Life is too short to be chasing DST bugs that crop up twice annually.

---

globals versus locals

 responseOnCall = requests.get(endpointoncall, headers=headers, params=params)

I am not excited to see headers and params being communicated as globals. Prefer assignment to local variables, or pass them into this function as parameters if you must.

This sort of coupling just leads to bugs and maintenance headaches down the road.

---

describe the return value

# get the overrides and userID to get the number later
def userid():
 ...
 ...
 return False
 else:
 # returns a list of dicts for overrides
 for override in overrides:
 save_override_info(override)
 return True

This is terrible. Re-write it or delete it.

The # comment said one thing, while the userid identifier said another, and there is no """docstring""" and no optional type annotation.

Consider annotating the return value:

def userid() -> bool:

In which case mypy would loudly complain about falling off the end which does a return None, clearly different from a boolean.

I don't understand what this predicate does. (And its name does not start with is_....) Maybe we're mostly executing it for logging side effects? In which case, document that. Whatever you do, invent a better name for this function, which clearly does not return a userid.

---

magic number

 userInfo['Override']['date_start'] = date_start[:16]
 userInfo['Override']['date_end'] = date_end[:16]
 ...
 return True

This code is too tricky. The "date" suggests we want Y-M-D. Yet what we get is Y-M-D H:M. At a minimum a # comment should explain that.

Also, why would we define a predicate that can only return a single value? Shouldn't the """docstring""" be explaining that we only evaluate this for side effects?

---

unmentioned globals

def phonenumber(userID, whatType):
 ...
 userInfo[whatType]['mobile'] = ...
 elif ... :
 userInfo[whatType]['email'] = ...

Ok, the globals situation is out of control here. Define a class and set self.foo instance attributes. Or pass in parameters. But this "all the world's a global variable!" nonsense has got to stop. Especially when there's no """docstring""" describing the contract. The # comment takes a stab at it, but it doesn't describe the behavior.

---

check exit status

 stdout, stderr = run.communicate()
 ...
 except subprocess.CalledProcessError as e:

Maybe that's what you want? But it would be more usual to call check_output(), so you'll notice any non-zero status values that might crop up.

It's not obvious to me that non-zero status would actually produce a CalledProcessError.

Wait for process to terminate and set the returncode attribute.

Yeah, I was kind of expecting to get an integer back in run, rather than an exception.

---

simplify regex

 pattern = r'^02\d{9}|02\d{8}$'

Prefer r'^02\d{8,9}$'

Also, you apparently intended:

 pattern = r'^(02\d{9}|02\d{8})$'

As written, you could match "02123456789trash" and also match "trash0212345678".

---

a boolean variable is already a boolean expression

 if override_needed is True and ...

Prefer if override_needed and ....

Further down, prefer not override_needed.

---

This codebase appears to achieve many of its design goals. It relies on poorly documented globals and lacks automated unit tests.

I would not be willing to delegate or accept maintenance tasks on this code in its current form.

• Alex Orbit: Thank you go much copied this to my wiki, and going to work on what you have said again thank you we use hashicorp vault at work, and Puppet gets keys from there, so as I'm looking at getting this installed on a server with puppet I may make this a template for Puppet to pull that API from vault, and not be in the script itself in git do you mind if I get back to this after I have made some changes in the coming week, sorry don't know 100% when I will get the time I may just do it after hours and again thank you so much for taking the time, have a wonderful end of your weekend
• Alex Orbit: hey sorry, didn't get time to make the changes I wanted to from your comment sorry, just got snowed under with other work, still going to make these changes im hoping this week