Updates a Slack User Group with People that are on call in PagerDuty (updated for slack_sdk after previous auth method was deprecated, changed user lookup on slack). Based on: https://gist.github.com/markddavidoff/863d77c351672345afa3fd465a970ad6

slack-pagerduty-oncall.py

"""
Script to update on-call groups based on pagerduty escalation policies
From: https://gist.github.com/markddavidoff/863d77c351672345afa3fd465a970ad6

Slack permissions required:
- Installer must be able to update user groups.
- usergroups:read
- usergroups:write
- users:read
- users:read.email

NOTE: Can also be deployed as a lambda function, more details on the link above. 
NOTE: Can also be executed with params from command line

TODO: 
[] - Handle slack api error on findByEmail (missing user)

"""
# !/usr/bin/env python

import json
import os
import logging

from urllib.parse import urlencode
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

logging.basicConfig(
    format='%(asctime)s %(levelname)-8s %(message)s',
    level=logging.INFO,
    datefmt='%Y-%m-%d %H:%M:%S')
log = logging.getLogger(__name__)

class SlackOnCall(object):
    # The Slack API token to use for authentication to the Slack WebAPI
    slack_token = None

    # The Pager Duty API token to use for authentication into the PagerDuty API
    pager_duty_token = None

    # The Slack @user-group to update (Default: oncall)
    slack_user_group_handle = 'oncall'

    # The maximum escalation level to add to the group
    # (eg. if escalation level = 2, then levels 1 and 2 will be a part of the group
    # but not any levels 3 and above.
    escalation_level = 2

    def __init__(self, slack_token, pager_duty_token,
                 slack_user_group_handle=slack_user_group_handle, log_level='INFO',
                 escalation_level=escalation_level, escalation_policy_ids=None):

        self.slack_token = slack_token
        self.pager_duty_token = pager_duty_token
        self.slack_user_group_handle = slack_user_group_handle
        self.escalation_level = int(escalation_level)
        if not escalation_policy_ids:
            self.escalation_policy_ids = []
        else:
            self.escalation_policy_ids = [s.strip() for s in escalation_policy_ids.split(',')]

        self._slack_user_group = None
        self._on_call_email_addresses = None
        self._all_slack_users = None

        log.setLevel(log_level)

    def run(self):
        """
        Gets user group information and on-call information then updates the
        on-call user group in slack to be the on-call users for escalation
        levels from input.
        """
        slack_users = self.slack_users_by_email(self.on_call_email_addresses)
        if not slack_users:
            log.warning('No Slack users found for email addresses: %s', ','.join(self.on_call_email_addresses))
            return

        slack_user_ids = [u['id'] for u in slack_users]

        if set(slack_user_ids) == set(self.slack_user_group['users']):
            log.info('User group %s already set to %s', self.slack_user_group_handle, slack_user_ids)
            return

        self.update_on_call(slack_users)
        log.info('Job Complete')

    @staticmethod
    def _make_request(url, body=None, headers={}):
        req = Request(url, body, headers )
        log.debug('Making request to %s', url)

        try:
            response = urlopen(req)
            body = response.read()
            try:
                body = json.loads(body)
                if 'error' in body:
                    msg = 'Error making request: {}'.format(body)
                    log.error(msg)
                    raise ValueError(msg)
                return body
            except ValueError:
                return body

        except HTTPError as e:
            log.error("Request failed: %d %s", e.code, e.reason)
        except URLError as e:
            log.error("Server connection failed: %s", e.reason)

    @property
    def slack_user_group(self):
        """
        :return: the Slack user group matching the slack_user_group_handle
        specified in the configuration
        """
        if self._slack_user_group is not None:
            return self._slack_user_group

        client = WebClient(token=self.slack_token)
        groups = client.usergroups_list(include_users=True)['usergroups']

        for group in groups:
            if group['handle'] == self.slack_user_group_handle:
                self._slack_user_group = group
                return group

        raise ValueError('No user groups found that match {}'.format(self.slack_user_group_handle))

    @property
    def on_call_email_addresses(self):
        """
        Hits the PagerDuty API and gets the on-call users from the escalation policies 
        and level from input and returns their email addresses
        :return: All on-call email addresses within the escalation bounds
        """
        if self._on_call_email_addresses is not None:
            return self._on_call_email_addresses

        log.info('Looking for users on EPs %s level %s and lower', self.escalation_policy_ids, self.escalation_level)

        params = {"escalation_policy_ids[]": self.escalation_policy_ids}
        url ='https://api.pagerduty.com/oncalls?time_zone=UTC&include%5B%5D=users&' + urlencode(params, True)
        on_call = self._make_request(url, headers={
            'Authorization': 'Token token=' + self.pager_duty_token,
            'Accept': 'application/vnd.pagerduty+json;version=2'
        })
        users = set()  # users can be in multiple schedule, this will de-dupe
        for escalation_policy in on_call['oncalls']:
            if escalation_policy['escalation_level'] <= self.escalation_level:
                users.add(escalation_policy['user']['email'])

        log.info('Found %d users on-call', len(users))
        log.debug(users)

        self._on_call_email_addresses = users
        return users

    def slack_users_by_email(self, emails):
        """
        Finds input slack users by their email address
        :param emails: List of email address to find users
        :return: List of Slack user objects found in :emails:
        """
        client = WebClient(token=self.slack_token)
        users = []
        for email in emails:
            user = client.users_lookupByEmail(email=email)
            log.debug('Found %s in slack', email)
            users.append(user["user"])

        return users

    def update_on_call(self, slack_users):
        """
        Updates the specified user-group
        :param slack_users: Slack users to modify the group with
        """
        user_ids = [u['id'] for u in slack_users]

        client = WebClient(token=self.slack_token)
        log.info('Updating user group %s from %s to %s',
                 self.slack_user_group_handle, self.slack_user_group['users'], user_ids)

        client.usergroups_users_update(usergroup=self.slack_user_group['id'], users=','.join(user_ids))


# def lambda_handler(*_):
#     """
#     Main entry point for AWS Lambda.

#     Variables can not be passed in to AWS Lambda, the configuration
#     parameters below are encrypted using AWS IAM Keys.
#     """

#     # Boto is always available in AWS lambda, but may not be available in
#     # standalone mode
#     import boto3

#     # To generate the encrypted values, go to AWS IAM Keys and Generate a key
#     # Then grant decryption using the key to the IAM Role used for your lambda
#     # function.
#     #
#     # Use the command `aws kms encrypt --key-id alias/<key-alias> --plaintext <value-to-encrypt>
#     # Put the encrypted value in the configuration dictionary below
#     config = {
#         'slack_token': os.environ['SLACK_API_TOKEN'],
#         'slack_user_group_handle': os.environ['SLACK_USER_GROUP_HANDLE'],
#         'pager_duty_token': os.environ['PAGERDUTY_API_TOKEN'],
#         'escalation_level': os.environ['ESCALATION_LEVEL'],
#         'escalation_policy_ids': os.environ['ESCALATION_POLICY_IDS']
#     }
#     return SlackOnCall(**config).run()

# def main():
#     """
#     Runs the Slack PagerDuty OnCall group updater as a standalone script
#     """
#     from argparse import ArgumentParser

#     parser = ArgumentParser(usage=main.__doc__)
#     parser.add_argument('-st', '--slack-token', required=True, dest='slack_token',
#                         help='Slack token to use for auth into the Slack WebAPI')
#     parser.add_argument('-su', '--slack-user-group', dest='slack_user_group_handle', default='oncall',
#                         help='Slack user group to add on-call users to. (Default: oncall)')
#     parser.add_argument('-pt', '--pager-duty-token', required=True, dest='pager_duty_token',
#                         help='PagerDuty token to use for auth into the PagerDuty API')
#     parser.add_argument('-el', '--max-escalation-level', dest='escalation_level', default=2, type=int,
#                         help='Max escalation level to add on-call users for group. (Default: 2)')
#     parser.add_argument('-ep', '--escalation-policy-ids', required=True, dest='escalation_policy_ids', default=[], type=str,
#                         help='List of escalation policies (Default: [])')
#     logging.basicConfig()
#     args = vars(parser.parse_args())
#     SlackOnCall(**args).run()

def main():
    """
    Runs the Slack PagerDuty OnCall group updater with inputs from env_vars
    """
    from argparse import ArgumentParser

    config = {
        'slack_token': os.environ['SLACK_API_TOKEN'],
        'slack_user_group_handle': os.environ.get('SLACK_USER_GROUP_HANDLE', "tr-oncall"),
        'pager_duty_token': os.environ['PAGERDUTY_API_TOKEN'],
        'escalation_level': os.environ.get('ESCALATION_LEVEL', 1),
        'escalation_policy_ids': os.environ['ESCALATION_POLICY_IDS']
    }
    return SlackOnCall(**config).run()

if __name__ == '__main__':
    main()

@tr-fteixeira you may also find: https://github.com/markddavidoff/slack-smart-alias interesting if you're using this the same way i was