pagerduty-oncall.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" List On-calls """

## Configuration ##

#secrets.json
#{
#  "pagerduty": {
#    "key": "apikey"
#  }
#}

#config-*.json // pull the user's phone numbers from their contact details
#{
#  "responders": [
#    {
#      "escalation_policy_ids": [ "POLICYID" ],
#      "schedule_ids": [ "SCHEDULEID" ]
#    }
#  ]
#}

#config-*-lcr.json // only provide the Live Call Routing number
#{
#  "responders": [
#    {
#      "escalation_policy_ids": [ "POLICYID" ],
#      "schedule_ids": [ "SCHEDULEID" ],
#      "lcr": "+xx xxx xxx xxxx,,y"
#    }
#  ]
#}



import json
import logging
from argparse import ArgumentParser

# Use the PagerDuty Python REST API Sessions library
# https://pagerduty.github.io/pdpyras/
# A local copy is included with this release
from pdpyras import APISession

SECRETS_FILE = "secrets.json"
CONFIG_FILE = "config.json"


def parse_args():
    """ Argument parsing is fun! """

    parser = ArgumentParser()
    parser.add_argument(
        '--config',
        help='Full path to config file.',
        metavar='path'
    )
    parser.add_argument(
        '--secrets',
        help='PagerDuty secrets file.',
        metavar='secrets'
    )
    parser.add_argument(
        '--verbose',
        help='Be more verbose in output.',
        action='store_true'
    )
    return parser.parse_args()


def main():
    """ Do all the things """

    logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
    args = parse_args()
    with open(args.config or CONFIG_FILE) as config:
        config = json.load(config)
    with open(args.secrets or SECRETS_FILE) as secrets:
        secrets = json.load(secrets)

    # instantiate a pdpyras object
    api = APISession(
        secrets.get('pagerduty').get('key'), # PagerDuty API token from secrets
        'PagerDuty' # Name for logging (optional)
        ## or, with default_from:
        #'PagerDuty', # Name for logging (optional)
        #'[email protected]' # default_from for request header - must be a valid PagerDuty user
        )

    for responder in config.get('responders'):
        # Generally speaking, at the first esclation level where these schedule(s) are mentioned,
        # there should be only one active on-call, regardless of time of day.
        # The expectation is that a level with split day/night or follow-the-sun responsibility will not overlap.

        logging.debug('Retrieving oncalls from PagerDuty API')

        # Poll for a single oncall, based on a schedule id and escalation policy id
        # The use of both Schedule and Escalation Policy limits scope
        oncalls = api.iter_all(
            'oncalls', # method
            {
                #"include[]": "users", # including users doesn't give us the contact details
                "schedule_ids[]": responder.get('schedule_ids'),
                "escalation_policy_ids[]": responder.get('escalation_policy_ids')
            } #params
        )

        if oncalls:
            for oncall in oncalls:
                # If we have a Live Call Routing number configured, just display it here
                if responder.get('lcr'):
                    ## directly print the result
                    # schedule - Live Call Routing: +xx xxxxxx,,x User Name until yyyy-mm-ddThh:mm:ssZ
                    print(u'`{}` - Live Call Routing: `{}` {} until {}'.format(
                        oncall.get('schedule').get('summary'),
                        responder.get('lcr'),
                        oncall.get('user').get('summary'),
                        oncall.get('end')
                    ))

                # We don't have an LCR configured, so find the user's phone number(s)
                else:
                    response = api.request(
                        'get', # requests type
                        '/users/{}'.format(oncall.get('user').get('id')), # get single user
                        params={"include[]": "contact_methods"} # include contact _details_
                    )
                    # prepare an empty phone list to populate - there could be more than one
                    phone = {}
                    if response.ok:
                        user = response.json()['user']
                        # loop through all of the contact methods, looking for phone numbers
                        for contact_method in user.get('contact_methods'):
                            logging.debug(u'Contact Method: %s %s',
                                        contact_method.get('type'),
                                        contact_method.get('label'))
                            # add phone numbers in the constructed format to the `phone` list
                            if 'phone' in contact_method['type']:
                                phone[contact_method.get('label')] = u'`{}: +{} {}`'.format(
                                    contact_method.get('label'),
                                    contact_method.get('country_code'),
                                    contact_method.get('address'))
                    # if no `contact_methods` of type `phone`
                    if not phone:
                        phone['EMPTY'] = 'NO PHONE ENTRIES FOUND'

                    ## print the result
                    # schedule - user name - Work: +xx xxxxxx, Mobile: +xx xxxxxx until yyyy-mm-ddThh:mm:ssZ
                    print(u'`{}` {} - {} until {}'.format(
                        oncall.get('schedule').get('summary'),
                        oncall.get('user').get('summary'),
                        ', '.join(phone.values()),
                        oncall.get('end')
                        )
                    )

                # In the event that multiple results are returned, the first result is provided
                break

        else:
            logging.critical('No oncalls returned')


if __name__ == '__main__':
    main()