Examples

Code examples can be found in the examples directory on GitHub

Also take a look at the unit tests in the tests directory.

Get all calling users

Source: calling_users.py

#!/usr/bin/env python
"""
Example script
Get all calling users within the org
"""

from dotenv import load_dotenv

from wxc_sdk import WebexSimpleApi

load_dotenv()

api = WebexSimpleApi()

# using wxc_sdk.people.PeopleApi.list to iterate over persons
# Parameter calling_data needs to be set to true to gat calling specific information
# calling users have the attribute location_id set
calling_users = [user for user in api.people.list(calling_data=True)
                 if user.location_id]
print(f'{len(calling_users)} users:')
print('\n'.join(user.display_name for user in calling_users))

Get all calling users (async variant)

Source: calling_users_async.py

#!/usr/bin/env python
"""
Example script
Get all calling users within the org using the (experimental) async API
"""
import asyncio
import time

from dotenv import load_dotenv

from wxc_sdk.as_api import AsWebexSimpleApi

load_dotenv()


async def get_calling_users():
    """
    Get details of all calling enabled users by:
    1) getting all calling users
    2) collecting all users that have a calling license
    3) getting details for all users
    """
    async with AsWebexSimpleApi(concurrent_requests=40) as api:
        print('Collecting calling licenses')
        calling_license_ids = set(lic.license_id for lic in await api.licenses.list()
                                  if lic.webex_calling)

        # get users with a calling license
        calling_users = [user async for user in api.people.list_gen()
                         if any(lic_id in calling_license_ids for lic_id in user.licenses)]
        print(f'{len(calling_users)} users:')
        print('\n'.join(user.display_name for user in calling_users))

        # get details for all users
        start = time.perf_counter()
        details = await asyncio.gather(*[api.people.details(person_id=user.person_id, calling_data=True)
                                         for user in calling_users])
        expired = time.perf_counter() - start
        print(f'Got details for {len(details)} users in {expired * 1000:.3f} ms')


asyncio.run(get_calling_users())

Get all users without phones

Source: users_wo_devices.py

#!/usr/bin/env python
"""
Get calling users without devices
"""
import asyncio
import logging
import os
from itertools import chain
from typing import Optional

from dotenv import load_dotenv

from wxc_sdk import Tokens
from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.common import UserType
from wxc_sdk.integration import Integration
from wxc_sdk.person_settings import PersonDevicesResponse
from wxc_sdk.scopes import parse_scopes


def env_path() -> str:
    """
    determine path for .env to load environment variables from; based on name of this file
    :return: .env file path
    """
    return os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.env')


def yml_path() -> str:
    """
    determine path of YML file to persist tokens
    :return: path to YML file
    :rtype: str
    """
    return os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.yml')


def build_integration() -> Integration:
    """
    read integration parameters from environment variables and create an integration
    :return: :class:`wxc_sdk.integration.Integration` instance
    """
    client_id = os.getenv('INTEGRATION_CLIENT_ID')
    client_secret = os.getenv('INTEGRATION_CLIENT_SECRET')
    scopes = parse_scopes(os.getenv('INTEGRATION_SCOPES'))
    redirect_url = 'http://localhost:6001/redirect'
    if not all((client_id, client_secret, scopes)):
        raise ValueError('failed to get integration parameters from environment')
    return Integration(client_id=client_id, client_secret=client_secret, scopes=scopes,
                       redirect_url=redirect_url)


def get_tokens() -> Optional[Tokens]:
    """
    Tokens are read from a YML file. If needed an OAuth flow is initiated.

    :return: tokens
    :rtype: :class:`wxc_sdk.tokens.Tokens`
    """

    integration = build_integration()
    tokens = integration.get_cached_tokens_from_yml(yml_path=yml_path())
    return tokens


async def main():
    # get environment variables from .env; required for integration parameters
    load_dotenv(env_path())

    # get tokens from cache or create a new set of tokens using the integration defined in .env
    tokens = get_tokens()

    async with AsWebexSimpleApi(tokens=tokens) as api:
        # get calling users
        calling_users = [user for user in await api.people.list(calling_data=True)
                         if user.location_id]

        # get device info for all users
        user_device_infos = await asyncio.gather(*[api.person_settings.devices(person_id=user.person_id)
                                                   for user in calling_users])
        user_device_infos: list[PersonDevicesResponse]
        users_wo_devices = [user for user, device_info in zip(calling_users, user_device_infos)
                            if not device_info.devices]

        # alternatively we can collect all device owners
        device_owner_ids = set(owner.owner_id
                               for device in chain.from_iterable(udi.devices for udi in user_device_infos)
                               if (owner := device.owner) and owner.owner_type == UserType.people)

        # ... and collect all other users (the ones not owning a device)
        users_not_owning_a_device = [user for user in calling_users
                                     if user.person_id not in device_owner_ids]

    users_wo_devices.sort(key=lambda u: u.display_name)
    print(f'{len(users_wo_devices)} users w/o devices:')
    print('\n'.join(f'{user.display_name} ({user.emails[0]})'
                    for user in users_wo_devices))

    print()
    users_not_owning_a_device.sort(key=lambda u: u.display_name)
    print(f'{len(users_not_owning_a_device)} users not owning a device:')
    print('\n'.join(f'{user.display_name} ({user.emails[0]})'
                    for user in users_not_owning_a_device))


if __name__ == '__main__':
    # enable DEBUG logging to a file; REST log shows all requests
    logging.basicConfig(filename=os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.log'),
                        filemode='w', level=logging.DEBUG, format='%(asctime)s %(threadName)s %(message)s')
    asyncio.run(main())

Default call forwarding settings for all users

This example start with the list of all calling users and then calls wxc_sdk.person_settings.forwarding.PersonForwardingApi.configure() for each user. To speed up things a ThreadPoolExecutor is used and all update operations are scheduled for execution by the thread pool.

Source: reset_call_forwarding.py

#!/usr/bin/env python
"""
Example script
Reset call forwarding to default for all users in the org
"""
import asyncio
import logging
import time

from dotenv import load_dotenv

from wxc_sdk.all_types import PersonForwardingSetting
from wxc_sdk.as_api import AsWebexSimpleApi


async def main():
    # load environment. SDk fetches the access token from WEBEX_ACCESS_TOKEN environment variable
    load_dotenv()

    logging.basicConfig(level=logging.INFO)

    # set to DEBUG to see the actual requests
    logging.getLogger('wxc_sdk.rest').setLevel(logging.INFO)

    async with AsWebexSimpleApi() as api:

        # get all calling users
        start = time.perf_counter_ns()
        calling_users = [user for user in await api.people.list(calling_data=True)
                         if user.location_id]
        print(f'Got {len(calling_users)} calling users in '
              f'{(time.perf_counter_ns() - start) / 1e6:.3f} ms')

        # set call forwarding to default for all users
        # default call forwarding settings
        forwarding = PersonForwardingSetting.default()

        # schedule update for each user and wait for completion
        start = time.perf_counter_ns()
        await asyncio.gather(*[api.person_settings.forwarding.configure(person_id=user.person_id,
                                                                        forwarding=forwarding)
                               for user in calling_users])
        print(f'Reset call forwarding to default for {len(calling_users)} users in '
              f'{(time.perf_counter_ns() - start) / 1e6:.3f} ms')

if __name__ == '__main__':
    asyncio.run(main())

Modify number of rings configuration for users read from CSV

Here we read a bunch of user email addresses from a CSV. The CSV as an ERROR column and we only want to consider users without errors.

For all these users a VM setting is updated and the results are written to a CSV for further processing.

Source: modify_voicemail.py

#!/usr/bin/env python
"""
Example Script
Modifying number of rings configuration in voicemail settings
Run -> python3 modify_voicemail.py modify_voicemail.csv
"""
import csv
import sys
import traceback
from concurrent.futures import ThreadPoolExecutor

from dotenv import load_dotenv

from wxc_sdk import WebexSimpleApi
from wxc_sdk.all_types import *

VOICEMAIL_SETTINGS_NUMBER_OF_RINGS = 6

# loading environment variables - use .env file for development
load_dotenv()


def update_vm_settings():
    """
    actually update VM settings for all users present in input CSV
    """
    api = WebexSimpleApi()
    final_report = []
    mail_ids = []
    # using wxc_sdk.people.PeopleApi.list to iterate over persons
    # Parameter calling_data needs to be set to true to gat calling specific information
    # calling users have the attribute location_id set
    calling_users = [user for user in api.people.list(calling_data=True)
                     if user.location_id]
    print(f'{len(calling_users)} users:')
    print('\n'.join(user.display_name for user in calling_users))

    # get CSV file name from command line
    with open(str(sys.argv[1]), 'r') as csv_file:
        reader = csv.DictReader(csv_file)
        # read all records from CSV. Ony consider records w/o error
        for col in reader:
            if not col['ERROR']:
                # collect email address from USERNAME column for further processing
                mail_ids.append(col['USERNAME'])
            else:
                final_report.append((col['USERNAME'], 'FAILED DUE TO ERROR REASON IN INPUT FILE'))

    # work on calling users that have an email address that we read from the CSV
    filteredUsers = [d for d in calling_users if d.emails[0] in mail_ids]

    print("\nCalling Users in CI  - Count ", len(calling_users))
    print("Mail IDs from input file after removing error columns - Count ", len(mail_ids))
    print("FilteredUsers Users -  Count", len(filteredUsers))

    def set_number_of_rings(user: Person):
        """
        Read VM config for a user
        :param user: user to update
        """
        try:
            # shortcut
            vm = api.person_settings.voicemail

            # Read Current configuration
            vm_settings = vm.read(person_id=user.person_id)
            print(f'\n Existing Configuration: {vm_settings} ')

            # Modify number of rings value
            vm_settings.send_unanswered_calls.number_of_rings = VOICEMAIL_SETTINGS_NUMBER_OF_RINGS
            vm.configure(person_id=user.person_id, settings=vm_settings)
            # Read configuration after changes
            vm_settings = vm.read(person_id=user.person_id)
            print(f'\n New Configuration: {vm_settings} ')
            final_report.append((user.display_name, 'SUCCESS'))
        except Exception as e:
            final_report.append((user.display_name, 'FAILURE'))
            print("type error: " + str(e))
            print(traceback.format_exc())
        return

    # Modify settings for the filtered users
    with ThreadPoolExecutor() as pool:
        list(pool.map(lambda user: set_number_of_rings(user),
                      filteredUsers))

    print(final_report)
    with open('output.csv', 'w') as f:
        write = csv.writer(f)
        write.writerow(["USERNAME", "STATUS"])
        write.writerows(final_report)


if __name__ == '__main__':
    update_vm_settings()

Holiday schedule w/ US national holidays for all US locations

This example uses the Calendarific API at https://calendarific.com/ to get a list of national US holidays and creates a “National Holidays” holiday schedule for all US locations with all these national holidays.

A rudimentary API implementation in calendarific.py is used for the requests to https://calendarific.com/. Calendarific APIs require all requests to be authenticated using an API key. You can sign up for a free account to get a free API account key which then is read from environment variable CALENDARIFIC_KEY.

Source: us_holidays.py

#!/usr/bin/env python
"""
Example script
Create a holiday schedule for all US locations with all national holidays
"""

import logging
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from datetime import date
from threading import Lock
from typing import List

from dotenv import load_dotenv

from calendarific import CalendarifiyApi, Holiday
from wxc_sdk import WebexSimpleApi
from wxc_sdk.locations import Location
from wxc_sdk.all_types import ScheduleType, Event, Schedule

log = logging.getLogger(__name__)

# a lock per location to protect observe_in_location()
location_locks: dict[str, Lock] = defaultdict(Lock)

# Use parallel threads for provisioning?
USE_THREADING = True

# True: delete holiday schedule instead of creating one
CLEAN_UP = False

# first and last year for which to create public holiday events
FIRST_YEAR = 2022
LAST_YEAR = 2024

LAST_YEAR = not CLEAN_UP and LAST_YEAR or FIRST_YEAR


def observe_in_location(*, api: WebexSimpleApi, location: Location, holidays: List[Holiday]):
    """
    create/update a "National Holiday" schedule in one location

    :param api: Webex api
    :type api: WebexSimpleApi
    :param location: location to work on
    :type location: Location
    :param holidays: list of holidays to observe
    :type holidays: List[Holiday]
    """
    # there should always only one thread messing with the holiday schedule of a location
    with location_locks[location.location_id]:
        year = holidays[0].date.year
        schedule_name = 'National Holidays'

        # shortcut
        ats = api.telephony.schedules

        # existing "National Holiday" schedule or None
        schedule = next((schedule
                         for schedule in ats.list(obj_id=location.location_id,
                                                  schedule_type=ScheduleType.holidays,
                                                  name=schedule_name)
                         if schedule.name == schedule_name),
                        None)
        if CLEAN_UP:
            if schedule:
                log.info(f'Delete schedule {schedule.name} in location {schedule.location_name}')
                ats.delete_schedule(obj_id=location.location_id,
                                    schedule_type=ScheduleType.holidays,
                                    schedule_id=schedule.schedule_id)
            return
        if schedule:
            # we need the details: list response doesn't have events
            schedule = ats.details(obj_id=location.location_id,
                                   schedule_type=ScheduleType.holidays,
                                   schedule_id=schedule.schedule_id)
        # create list of desired schedule entries
        #   * one per holiday
        #   * only future holidays
        #   * not on a Sunday
        today = date.today()
        events = [Event(name=f'{holiday.name} {holiday.date.year}',
                        start_date=holiday.date,
                        end_date=holiday.date,
                        all_day_enabled=True)
                  for holiday in holidays
                  if holiday.date >= today and holiday.date.weekday() != 6]

        if not schedule:
            # create new schedule
            log.debug(f'observe_in_location({location.name}, {year}): no existing schedule')
            if not events:
                log.info(f'observe_in_location({location.name}, {year}): no existing schedule, no events, done')
                return
            schedule = Schedule(name=schedule_name,
                                schedule_type=ScheduleType.holidays,
                                events=events)
            log.debug(
                f'observe_in_location({location.name}, {year}): creating schedule "{schedule_name}" with {len(events)} '
                f'events')
            schedule_id = ats.create(obj_id=location.location_id, schedule=schedule)
            log.info(f'observe_in_location({location.name}, {year}): new schedule id: {schedule_id}, done')
            return

        # update existing schedule
        with ThreadPoolExecutor() as pool:
            # delete existing events in the past
            to_delete = [event
                         for event in schedule.events
                         if event.start_date < today]
            if to_delete:
                log.debug(f'observe_in_location({location.name}, {year}): deleting {len(to_delete)} outdated events')
                if USE_THREADING:
                    list(pool.map(
                        lambda event: ats.event_delete(obj_id=location.location_id,
                                                       schedule_type=ScheduleType.holidays,
                                                       schedule_id=schedule.schedule_id,
                                                       event_id=event.event_id),
                        to_delete))
                else:
                    for event in to_delete:
                        ats.event_delete(obj_id=location.location_id,
                                         schedule_type=ScheduleType.holidays,
                                         schedule_id=schedule.schedule_id,
                                         event_id=event.event_id)

            # add events which don't exist yet
            existing_dates = set(event.start_date
                                 for event in schedule.events)
            to_add = [event
                      for event in events
                      if event.start_date not in existing_dates]
            if not to_add:
                log.info(f'observe_in_location({location.name}, {year}): no events to add, done.')
                return
            log.debug(f'observe_in_location({location.name}, {year}): creating {len(to_add)} new events.')
            if USE_THREADING:
                list(pool.map(
                    lambda event: ats.event_create(
                        obj_id=location.location_id,
                        schedule_type=ScheduleType.holidays,
                        schedule_id=schedule.schedule_id,
                        event=event),
                    to_add))
            else:
                for event in to_add:
                    ats.event_create(
                        obj_id=location.location_id,
                        schedule_type=ScheduleType.holidays,
                        schedule_id=schedule.schedule_id,
                        event=event)
        log.info(f'observe_in_location({location.name}, {year}): done.')
    return


def observe_national_holidays(*, api: WebexSimpleApi, locations: List[Location],
                              year: int = None):
    """
    US national holidays for given locations

    :param api: Webex api
    :type api: WebexSimpleApi
    :param locations: list of locations in which US national holidays should be observed
    :type locations: List[Location]
    :param year: year for national holidays. Default: current year
    :type year: int
    """
    # default: this year
    year = year or date.today().year

    # get national holidays for specified year
    holidays = CalendarifiyApi().holidays(country='US', year=year, holiday_type='national')

    # update holiday schedule for each location
    with ThreadPoolExecutor() as pool:
        if USE_THREADING:
            list(pool.map(
                lambda location: observe_in_location(api=api, location=location, holidays=holidays),
                locations))
        else:
            for location in locations:
                observe_in_location(api=api, location=location, holidays=holidays)
    return


if __name__ == '__main__':
    # read dotenv which has some environment variables like Webex API token and Calendarify
    # API key.
    load_dotenv()

    # enable logging
    logging.basicConfig(level=logging.DEBUG,
                        format='%(asctime)s %(levelname)s %(threadName)s %(name)s: %(message)s')
    logging.getLogger('urllib3').setLevel(logging.INFO)
    logging.getLogger('wxc_sdk.rest').setLevel(logging.INFO)

    # the actual action
    with WebexSimpleApi(concurrent_requests=5) as wx_api:
        # get all US locations
        log.info('Getting locations...')
        us_locations = [location
                        for location in wx_api.locations.list()
                        if location.address.country == 'US']

        # set up location locks
        # location_locks is a defaultdict -> accessing with all potential keys creates the locks
        list(location_locks[loc.location_id] for loc in us_locations)

        # create national holiday schedule for given year(s) and locations
        if USE_THREADING:
            with ThreadPoolExecutor() as pool:
                list(pool.map(
                    lambda year: observe_national_holidays(api=wx_api, year=year, locations=us_locations),
                    range(FIRST_YEAR, LAST_YEAR + 1)))
        else:
            for year in range(FIRST_YEAR, LAST_YEAR + 1):
                observe_national_holidays(api=wx_api, year=year, locations=us_locations)

Holiday schedule w/ US national holidays for all US locations (async variant)

This example uses the Calendarific API at https://calendarific.com/ to get a list of national US holidays and creates a “National Holidays” holiday schedule for all US locations with all these national holidays.

A rudimentary API implementation in calendarific.py is used for the requests to https://calendarific.com/. Calendarific APIs require all requests to be authenticated using an API key. You can sign up for a free account to get a free API account key which then is read from environment variable CALENDARIFIC_KEY.

Source: us_holidays_async.py

#!/usr/bin/env python
"""
Example script
Create a holiday schedule for all US locations with all national holidays

Using the asyc SDK variant
"""
import asyncio
import functools
import logging
from collections import defaultdict
from datetime import date
from typing import List

from dotenv import load_dotenv

from calendarific import CalendarifiyApi, Holiday
from wxc_sdk.all_types import ScheduleType, Event, Schedule
from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.locations import Location

log = logging.getLogger(__name__)

# a lock per location to protect observe_in_location()
location_locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)

# Use parallel tasks for provisioning?
USE_TASKS = True

# True: delete holiday schedule instead of creating one
CLEAN_UP = False

# first and last year for which to create public holiday events
FIRST_YEAR = 2022
LAST_YEAR = 2024

LAST_YEAR = not CLEAN_UP and LAST_YEAR or FIRST_YEAR


async def observe_in_location(*, api: AsWebexSimpleApi, location: Location, holidays: List[Holiday]):
    """
    create/update a "National Holiday" schedule in one location

    :param api: Webex api
    :type api: WebexSimpleApi
    :param location: location to work on
    :type location: Location
    :param holidays: list of holidays to observe
    :type holidays: List[Holiday]
    """
    # there should always only one thread messing with the holiday schedule of a location
    async with location_locks[location.location_id]:
        year = holidays[0].date.year
        schedule_name = 'National Holidays'

        # shortcut
        ats = api.telephony.schedules

        # existing "National Holiday" schedule or None
        schedule = next((schedule
                         for schedule in await ats.list(obj_id=location.location_id,
                                                        schedule_type=ScheduleType.holidays,
                                                        name=schedule_name)
                         if schedule.name == schedule_name),
                        None)
        if CLEAN_UP:
            if schedule:
                log.info(f'Delete schedule {schedule.name} in location {schedule.location_name}')
                await ats.delete_schedule(obj_id=location.location_id,
                                          schedule_type=ScheduleType.holidays,
                                          schedule_id=schedule.schedule_id)
            return
        if schedule:
            # we need the details: list response doesn't have events
            schedule = await ats.details(obj_id=location.location_id,
                                         schedule_type=ScheduleType.holidays,
                                         schedule_id=schedule.schedule_id)
        # create list of desired schedule entries
        #   * one per holiday
        #   * only future holidays
        #   * not on a Sunday
        today = date.today()
        events = [Event(name=f'{holiday.name} {holiday.date.year}',
                        start_date=holiday.date,
                        end_date=holiday.date,
                        all_day_enabled=True)
                  for holiday in holidays
                  if holiday.date >= today and holiday.date.weekday() != 6]

        if not schedule:
            # create new schedule
            log.debug(f'observe_in_location({location.name}, {year}): no existing schedule')
            if not events:
                log.info(f'observe_in_location({location.name}, {year}): no existing schedule, no events, done')
                return
            schedule = Schedule(name=schedule_name,
                                schedule_type=ScheduleType.holidays,
                                events=events)
            log.debug(
                f'observe_in_location({location.name}, {year}): creating schedule "{schedule_name}" with {len(events)} '
                f'events')
            schedule_id = await ats.create(obj_id=location.location_id, schedule=schedule)
            log.info(f'observe_in_location({location.name}, {year}): new schedule id: {schedule_id}, done')
            return

        # update existing schedule
        # delete existing events in the past
        to_delete = [event
                     for event in schedule.events
                     if event.start_date < today]
        if to_delete:
            log.debug(f'observe_in_location({location.name}, {year}): deleting {len(to_delete)} outdated events')
            if USE_TASKS:
                await asyncio.gather(*[ats.event_delete(obj_id=location.location_id,
                                                        schedule_type=ScheduleType.holidays,
                                                        schedule_id=schedule.schedule_id,
                                                        event_id=event.event_id)
                                       for event in to_delete])
            else:
                for event in to_delete:
                    await ats.event_delete(obj_id=location.location_id,
                                           schedule_type=ScheduleType.holidays,
                                           schedule_id=schedule.schedule_id,
                                           event_id=event.event_id)

        # add events which don't exist yet
        existing_dates = set(event.start_date
                             for event in schedule.events)
        to_add = [event
                  for event in events
                  if event.start_date not in existing_dates]
        if not to_add:
            log.info(f'observe_in_location({location.name}, {year}): no events to add, done.')
            return
        log.debug(f'observe_in_location({location.name}, {year}): creating {len(to_add)} new events.')
        if USE_TASKS:
            await asyncio.gather(*[ats.event_create(obj_id=location.location_id,
                                                    schedule_type=ScheduleType.holidays,
                                                    schedule_id=schedule.schedule_id,
                                                    event=event)
                                   for event in to_add])
        else:
            for event in to_add:
                await ats.event_create(obj_id=location.location_id,
                                       schedule_type=ScheduleType.holidays,
                                       schedule_id=schedule.schedule_id,
                                       event=event)
        log.info(f'observe_in_location({location.name}, {year}): done.')
    return


async def observe_national_holidays(*, api: AsWebexSimpleApi, locations: List[Location],
                                    year: int = None):
    """
    US national holidays for given locations

    :param api: Webex api
    :type api: WebexSimpleApi
    :param locations: list of locations in which US national holidays should be observed
    :type locations: List[Location]
    :param year: year for national holidays. Default: current year
    :type year: int
    """
    # default: this year
    year = year or date.today().year

    # get national holidays for specified year
    loop = asyncio.get_running_loop()
    # avoid sync all:
    # holidays = CalendarifiyApi().holidays(country='US', year=year, holiday_type='national')
    holidays = await loop.run_in_executor(None, functools.partial(CalendarifiyApi().holidays,
                                                                  country='US', year=year, holiday_type='national'))

    # update holiday schedule for each location
    if USE_TASKS:
        await asyncio.gather(*[observe_in_location(api=api, location=location, holidays=holidays)
                               for location in locations])
    else:
        for location in locations:
            await observe_in_location(api=api, location=location, holidays=holidays)
    return


if __name__ == '__main__':
    # read dotenv which has some environment variables like Webex API token and Calendarify
    # API key.
    load_dotenv()

    # enable logging
    logging.basicConfig(level=logging.DEBUG,
                        format='%(asctime)s %(levelname)s %(threadName)s %(name)s: %(message)s')
    logging.getLogger('urllib3').setLevel(logging.INFO)
    logging.getLogger('wxc_sdk.as_rest').setLevel(logging.INFO)

    # the actual action
    async def do_provision():

        async with AsWebexSimpleApi(concurrent_requests=5) as wx_api:
            # get all US locations
            log.info('Getting locations...')
            us_locations = [location
                            for location in await wx_api.locations.list()
                            if location.address.country == 'US']

            # set up location locks
            # location_locks is a defaultdict -> accessing with all potential keys creates the locks
            list(location_locks[loc.location_id] for loc in us_locations)

            # create national holiday schedule for given year(s) and locations
            if USE_TASKS:
                await asyncio.gather(*[observe_national_holidays(api=wx_api, year=year, locations=us_locations)
                                       for year in range(FIRST_YEAR, LAST_YEAR + 1)])
            else:
                for year in range(FIRST_YEAR, LAST_YEAR + 1):
                    await observe_national_holidays(api=wx_api, year=year, locations=us_locations)

    asyncio.run(do_provision())

Persist tokens and obtain new tokens interactively

A typical problem specifically when creating CLI scripts is how to obtain valid access tokens for the API operations. If your code wants to invoke Webex REST APIs on behalf of a user then an integration is needed. The concepts of integrations are explained at the “Integrations” page on developer.cisco.com.

This example code shows how an OAUth Grant flow for an integration can be initiated from a script by using the Python webbrowser module and calling the open() method with the authorization URL of a given integration to open that URL in the system web browser. The user can then authenticate and grant access to the integration. In the last step of a successful authorization flow the web browser is redirected to the redirect_url of the integration.

The example code starts a primitive web server serving GET requests to http://localhost:6001/redirect. This URL has to be the redirect URL of the integration you create under My Webex Apps on developer.webex.com.

The sample script reads the integration parameters from environment variables (TOKEN_INTEGRATION_CLIENT_ID, TOKEN_INTEGRATION_CLIENT_SECRET, TOKEN_INTEGRATION_CLIENT_SCOPES). These variables can also be defined in get_tokens.env in the current directory:

# rename this to get_tokens.env and set values

# client ID for integration to be used.
TOKEN_INTEGRATION_CLIENT_ID=

# client secret of integration
TOKEN_INTEGRATION_CLIENT_SECRET=

# scopes to request. Use these scopes when creating the integration at developer.webex.com
# scopes can be in any form supported by wxc_sdk.scopes.parse_scopes()
TOKEN_INTEGRATION_CLIENT_SCOPES="spark:calls_write spark:kms spark:calls_read spark-admin:telephony_config_read spark:people_read"

The sample code persists the tokens in get_tokens.yml in the current directory. On startup the sample code tries to read tokens from that file. If needed a new access token is obtained using the refresh token.

An OAuth flow is only initiated if no (valid) tokens could be read from get_tokens.yml

Source: get_tokens.py

#!/usr/bin/env python
"""
Example script
read tokens from file or interactively obtain token by starting a local web server and open the authorization URL in
the local web browser
"""
import logging
import os
from typing import Optional

from dotenv import load_dotenv

from wxc_sdk import WebexSimpleApi
from wxc_sdk.integration import Integration
from wxc_sdk.scopes import parse_scopes
from wxc_sdk.tokens import Tokens

log = logging.getLogger(__name__)


def env_path() -> str:
    """
    determine path for .env to load environment variables from

    :return: .env file path
    """
    return os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.env')


def yml_path() -> str:
    """
    determine path of YML file to persist tokens

    :return: path to YML file
    :rtype: str
    """
    return os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.yml')


def build_integration() -> Integration:
    """
    read integration parameters from environment variables and create an integration

    :return: :class:`wxc_sdk.integration.Integration` instance
    """
    client_id = os.getenv('TOKEN_INTEGRATION_CLIENT_ID')
    client_secret = os.getenv('TOKEN_INTEGRATION_CLIENT_SECRET')
    scopes = parse_scopes(os.getenv('TOKEN_INTEGRATION_CLIENT_SCOPES'))
    redirect_url = 'http://localhost:6001/redirect'
    if not all((client_id, client_secret, scopes)):
        raise ValueError('failed to get integration parameters from environment')
    return Integration(client_id=client_id, client_secret=client_secret, scopes=scopes,
                       redirect_url=redirect_url)


def get_tokens() -> Optional[Tokens]:
    """

    Tokens are read from a YML file. If needed an OAuth flow is initiated.

    :return: tokens
    :rtype: :class:`wxc_sdk.tokens.Tokens`
    """

    integration = build_integration()
    tokens = integration.get_cached_tokens_from_yml(yml_path=yml_path())
    return tokens


logging.basicConfig(level=logging.DEBUG)

# load environment variables from .env
path = env_path()
log.info(f'reading {path}')
load_dotenv(env_path())

tokens = get_tokens()

# use the tokens to get identity of authenticated user
api = WebexSimpleApi(tokens=tokens)
me = api.people.me()
print(f'authenticated as {me.display_name} ({me.emails[0]}) ')

Read/update call intercept settings of a user

usage: call_intercept.py [-h] [–token TOKEN] user_email [{on,off}]

positional arguments:

user_email email address of user

{on,off} operation to apply

options:

-h, –help show this help message and exit

–token TOKEN admin access token to use

The script uses the access token passed via the CLI, reads one from the WEBEX_ACCESS_TOKEN environment variable or obtains tokens via an OAuth flow. For the last option the integration parameters are read from environment variables which can be set in a .env file

Source: call_intercept.py

#!/usr/bin/env python
"""
Script to read/update call intercept settings of a calling user.

The script uses the access token passed via the CLI, reads one from the WEBEX_ACCESS_TOKEN environment variable or
obtains tokens via an OAuth flow.

    usage: call_intercept.py [-h] [--token TOKEN] user_email [{on,off}]

    positional arguments:
      user_email     email address of user
      {on,off}       operation to apply

    options:
      -h, --help     show this help message and exit
      --token TOKEN  admin access token to use
"""
import argparse
import logging
import os
import re
import sys
from json import loads
from typing import Optional

from dotenv import load_dotenv
from wxc_sdk import WebexSimpleApi
from wxc_sdk.integration import Integration
from wxc_sdk.person_settings.call_intercept import InterceptSetting
from wxc_sdk.rest import RestError
from wxc_sdk.scopes import parse_scopes
from wxc_sdk.tokens import Tokens
from yaml import safe_dump, safe_load

log = logging.getLogger(__name__)


def env_path() -> str:
    """
    determine path for .env to load environment variables from

    :return: .env file path
    """
    return os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.env')


def yml_path() -> str:
    """
    determine path of YML file to persist tokens

    :return: path to YML file
    :rtype: str
    """
    return os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.yml')


def build_integration() -> Integration:
    """
    read integration parameters from environment variables and create an integration

    :return: :class:`wxc_sdk.integration.Integration` instance
    """
    client_id = os.getenv('TOKEN_INTEGRATION_CLIENT_ID')
    client_secret = os.getenv('TOKEN_INTEGRATION_CLIENT_SECRET')
    scopes = os.getenv('TOKEN_INTEGRATION_CLIENT_SCOPES')
    if scopes:
        scopes = parse_scopes(scopes)
    if not all((client_id, client_secret, scopes)):
        raise ValueError('failed to get integration parameters from environment')
    redirect_url = 'http://localhost:6001/redirect'
    return Integration(client_id=client_id, client_secret=client_secret, scopes=scopes,
                       redirect_url=redirect_url)


def get_tokens() -> Optional[Tokens]:
    """

    Tokens are read from a YML file. If needed an OAuth flow is initiated.

    :return: tokens
    :rtype: :class:`wxc_sdk.tokens.Tokens`
    """

    def write_tokens(tokens_to_cache: Tokens):
        with open(yml_path(), mode='w') as f:
            safe_dump(loads(tokens_to_cache.json()), f)
        return

    def read_tokens() -> Optional[Tokens]:
        try:
            with open(yml_path(), mode='r') as f:
                data = safe_load(f)
                tokens_read = Tokens.parse_obj(data)
        except Exception as e:
            log.info(f'failed to read tokens from file: {e}')
            tokens_read = None
        return tokens_read

    integration = build_integration()
    tokens = integration.get_cached_tokens(read_from_cache=read_tokens,
                                           write_to_cache=write_tokens)
    return tokens


RE_EMAIL = re.compile(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$")


def email_type(value):
    if not RE_EMAIL.match(value):
        raise argparse.ArgumentTypeError(f"'{value}' is not a valid email")
    return value


def main():
    """
    where the magic happens
    """
    # read .env file with the settings for the integration to be used to obtain tokens
    load_dotenv(env_path())

    parser = argparse.ArgumentParser()
    parser.add_argument('user_email', type=email_type, help='email address of user')
    parser.add_argument('on_off', choices=['on', 'off'], nargs='?', help='operation to apply')
    parser.add_argument('--token', type=str, required=False, help='admin access token to use')
    args = parser.parse_args()

    if args.token:
        tokens = args.token
    elif (tokens := os.getenv('WEBEX_ACCESS_TOKEN')) is None:
        tokens = get_tokens()

    if not tokens:
        print('Failed to get tokens', file=sys.stderr)
        exit(1)

    # set level to DEBUG to see debug of REST requests
    logging.basicConfig(level=(gt := getattr(sys, 'gettrace', None)) and gt() and logging.DEBUG or logging.INFO)

    with WebexSimpleApi(tokens=tokens) as api:
        # get user
        email = args.user_email.lower()
        user = next((user
                     for user in api.people.list(email=email)
                     if user.emails[0] == email), None)
        if user is None:
            print(f'User "{email}" not found', file=sys.stderr)
            exit(1)

        # display call intercept status
        try:
            intercept = api.person_settings.call_intercept.read(person_id=user.person_id)
        except RestError as e:
            print(f'Failed to read call intercept settings: {e.response.status_code}, {e.description}')
            exit(1)

        print('on' if intercept.enabled else 'off')
        if args.on_off:
            # action: turn on/off
            intercept = InterceptSetting.default()
            intercept.enabled = args.on_off == 'on'
            try:
                api.person_settings.call_intercept.configure(person_id=user.person_id,
                                                             intercept=intercept)
            except RestError as e:
                print(f'Failed to update call intercept settings: {e.response.status_code}, {e.description}')
                exit(1)

            # read call intercept again
            try:
                intercept = api.person_settings.call_intercept.read(person_id=user.person_id)
            except RestError as e:
                print(f'Failed to read call intercept settings: {e.response.status_code}, {e.description}')
                exit(1)

            # display state after update
            print(f"set to {'on' if intercept.enabled else 'off'}")

    exit(0)


if __name__ == '__main__':
    main()

Read/update call queue agent join states

usage: queue_helper.py [-h] [–location LOCATION [LOCATION …]]

[–queue QUEUE [QUEUE …]]

[–join JOIN_AGENT [JOIN_AGENT …]]

[–unjoin UNJOIN_AGENT [UNJOIN_AGENT …]]

[–remove REMOVE_USER [REMOVE_USER …]]

[–add ADD_USER [ADD_USER …]] [–dryrun]

[–token TOKEN]

Modify call queue settings from the CLI

optional arguments:

-h, –help show this help message and exit

–location LOCATION [LOCATION …], -l LOCATION [LOCATION …]

name of location to work on. If missing then work on

all locations.

–queue QUEUE [QUEUE …], -q QUEUE [QUEUE …]

name(s) of queue(s) to operate on. If missing then

work on all queues in location.

–join JOIN_AGENT [JOIN_AGENT …], -j JOIN_AGENT [JOIN_AGENT …]

Join given user(s) on given queue(s). Can be “all” to

act on all agents.

–unjoin UNJOIN_AGENT [UNJOIN_AGENT …], -u UNJOIN_AGENT [UNJOIN_AGENT …]

Unjoin given agent(s) from given queue(s). Can be

“all” to act on all agents.

–remove REMOVE_USER [REMOVE_USER …], -r REMOVE_USER [REMOVE_USER …]

Remove given agent from given queue(s). Can be “all”

to act on all agents.

–add ADD_USER [ADD_USER …], -a ADD_USER [ADD_USER …]

Add given users to given queue(s).

–dryrun, -d Dry run; don’t apply any changes

–token TOKEN admin access token to use

The script uses the access token passed via the CLI, reads one from the WEBEX_ACCESS_TOKEN environment variable or obtains tokens via an OAuth flow. For the last option the integration parameters are read from environment variables which can be set in a queue_helper.env file in the current directory.

Source: queue_helper.py

#!/usr/bin/env python
"""
usage: queue_helper.py [-h] [--location LOCATION [LOCATION ...]] [--queue QUEUE [QUEUE ...]]
                       [--join JOIN_AGENT [JOIN_AGENT ...]] [--unjoin UNJOIN_AGENT [UNJOIN_AGENT ...]]
                       [--remove REMOVE_USER [REMOVE_USER ...]] [--add ADD_USER [ADD_USER ...]] [--dryrun]
                       [--token TOKEN]

Modify call queue settings from the CLI

optional arguments:
  -h, --help            show this help message and exit
  --location LOCATION [LOCATION ...], -l LOCATION [LOCATION ...]
                        name of location to work on. If missing then work on all locations.
  --queue QUEUE [QUEUE ...], -q QUEUE [QUEUE ...]
                        name(s) of queue(s) to operate on. If missing then work on all queues in location.
  --join JOIN_AGENT [JOIN_AGENT ...], -j JOIN_AGENT [JOIN_AGENT ...]
                        Join given user(s) on given queue(s). Can be "all" to act on all agents.
  --unjoin UNJOIN_AGENT [UNJOIN_AGENT ...], -u UNJOIN_AGENT [UNJOIN_AGENT ...]
                        Unjoin given agent(s) from given queue(s). Can be "all" to act on all agents.
  --remove REMOVE_USER [REMOVE_USER ...], -r REMOVE_USER [REMOVE_USER ...]
                        Remove given agent from given queue(s). Can be "all" to act on all agents.
  --add ADD_USER [ADD_USER ...], -a ADD_USER [ADD_USER ...]
                        Add given users to given queue(s).
  --dryrun, -d          Dry run; don't apply any changes
  --token TOKEN         admin access token to use
"""
import asyncio
import logging
import os
import sys
from argparse import ArgumentParser
from collections.abc import AsyncGenerator
from typing import Optional

from dotenv import load_dotenv

from wxc_sdk import Tokens
from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.common import UserType
from wxc_sdk.integration import Integration
from wxc_sdk.people import Person
from wxc_sdk.scopes import parse_scopes
from wxc_sdk.telephony.callqueue import CallQueue
from wxc_sdk.telephony.hg_and_cq import Agent


def agent_name(agent: Agent) -> str:
    return f'{agent.first_name} {agent.last_name}'


def env_path() -> str:
    """
    determine path for .env to load environment variables from; based on name of this file
    :return: .env file path
    """
    return os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.env')


def yml_path() -> str:
    """
    determine path of YML file to persist tokens
    :return: path to YML file
    :rtype: str
    """
    return os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.yml')


def build_integration() -> Integration:
    """
    read integration parameters from environment variables and create an integration
    :return: :class:`wxc_sdk.integration.Integration` instance
    """
    client_id = os.getenv('INTEGRATION_CLIENT_ID')
    client_secret = os.getenv('INTEGRATION_CLIENT_SECRET')
    scopes = parse_scopes(os.getenv('INTEGRATION_SCOPES'))
    redirect_url = 'http://localhost:6001/redirect'
    if not all((client_id, client_secret, scopes)):
        raise ValueError('failed to get integration parameters from environment')
    return Integration(client_id=client_id, client_secret=client_secret, scopes=scopes,
                       redirect_url=redirect_url)


def get_tokens() -> Optional[Tokens]:
    """
    Tokens are read from a YML file. If needed an OAuth flow is initiated.

    :return: tokens
    :rtype: :class:`wxc_sdk.tokens.Tokens`
    """

    integration = build_integration()
    tokens = integration.get_cached_tokens_from_yml(yml_path=yml_path())
    return tokens


async def main():
    async def act_on_queue(queue: CallQueue):
        """
        Act on a single queue
        """
        # we need the queue details b/c the queue instance passed as parameter is from a list() call
        # ... and thus is missing all the details like agents
        details = await api.telephony.callqueue.details(location_id=queue.location_id, queue_id=queue.id)
        agent_names = set(map(agent_name, details.agents))

        def notify(message: str) -> str:
            """
            an action notification with queue information
            """
            return f'queue "{details.name:{queue_len}}" in "{queue.location_name:{location_len}}": {message}'

        def validate_agents(names: list[str], operation: str) -> list[str]:
            """
            check if all names in given list exist as agents on current queue
            """
            if 'all' in names:
                return set(agent_names)

            not_found = [name for name in names if name not in agent_names]
            if not_found:
                print('\n'.join(notify(f'{name} not found for {operation}"')
                                for name in not_found),
                      file=sys.stderr)
            return set(name for name in names if name not in set(not_found))

        # validate list of names or join, unjoin, and remove against actual list of agents
        to_join = validate_agents(join_agents, 'join')
        to_unjoin = validate_agents(unjoin_agents, 'unjoin')
        to_remove = validate_agents(remove_users, 'remove')

        # check for agents we are asked to add but which already exist as agents on the queue
        existing_agent_ids = set(agent.agent_id for agent in details.agents)
        agent_exists = [agent_name(user) for user in add_users
                        if user.person_id in existing_agent_ids]
        if agent_exists:
            print('\n'.join(notify(f'{name} already is agent')
                            for name in agent_exists),
                  file=sys.stderr)
            # reduced set of users to add
            to_add = [user for user in add_users
                      if agent_name(user) not in set(agent_exists)]
        else:
            # ...  or add all users
            to_add = add_users

        # the updated list of agents for the current queue
        new_agents = []

        # do we actually need an update?
        update_needed = False

        # create copy of each agent instance; we don't want to update the original agent objects
        # to make sure that details still holds the state before any update
        agents = [agent.copy(deep=True) for agent in details.agents]

        # iterate through the existing agents and see if we have to apply any change
        for agent in agents:
            name = agent_name(agent)
            # do we have to take action to join this agent?
            if name in to_join and not agent.join_enabled:
                print(notify(f'{name}, join'))
                update_needed = True
                agent.join_enabled = True
            # do we have to take action to unjoin this agent?
            if name in to_unjoin and agent.join_enabled:
                print(notify(f'{name}, unjoin'))
                update_needed = True
                agent.join_enabled = False
            # do we have to remove this agent?
            if name in to_remove:
                print(notify(f'{name}, remove'))
                update_needed = True
                # skip to next agent; so that we don't add this agent to the updated list of agents
                continue
            new_agents.append(agent)

        # add new agents
        new_agents.extend(Agent(agent_id=user.person_id, user_type=UserType.people)
                          for user in to_add)

        # update the queue
        if (update_needed or to_add) and not args.dryrun:
            # simplified update: we only messed with the agents
            update = CallQueue(agents=new_agents)
            await api.telephony.callqueue.update(location_id=queue.location_id, queue_id=queue.id,
                                                 update=update)
            print(notify('queue updated'))
            # and get details after the update
            details = await api.telephony.callqueue.details(location_id=queue.location_id, queue_id=queue.id)
            print(notify('got details after update'))

        # print summary
        print(f'queue "{queue.name:{queue_len}}" in "{queue.location_name}"')
        print(f'  phone number: {details.phone_number}')
        print(f'  extension: {details.extension}')
        print('  agents')
        if details.agents:
            name_len = max(map(len, map(agent_name, details.agents)))
            for agent in details.agents:
                print(f'    {agent_name(agent):{name_len}}: {"not " if not agent.join_enabled else ""}joined')
        return

    async def validate_users(user_names: list[str]) -> AsyncGenerator[Person, None, None]:
        """
        Validate list of names of users to be added and yield a Person instance for each one
        """
        # search for all names in parallel
        lists: list[list[Person]] = await asyncio.gather(
            *[api.people.list(display_name=name) for name in user_names], return_exceptions=True)
        for name, user_list in zip(user_names, lists):
            if isinstance(user_list, Exception):
                user = None
            else:
                user = next((u for u in user_list if name == agent_name(u)), None)
            if user is None:
                print(f'user "{name}" not found', file=sys.stderr)
                continue
            yield user
        return

    # parse command line
    parser = ArgumentParser(description='Modify call queue settings from the CLI')
    parser.add_argument('--location', '-l', type=str, required=False, nargs='+',
                        help='name of location to work on. If missing then work on all locations.')

    parser.add_argument('--queue', '-q', type=str, required=False, nargs='+',
                        help='name(s) of queue(s) to operate on. If missing then work on all queues in location.')

    parser.add_argument('--join', '-j', type=str, required=False, nargs='+', dest='join_agent',
                        help='Join given user(s) on given queue(s). Can be "all" to act on all agents.')

    parser.add_argument('--unjoin', '-u', type=str, required=False, nargs='+', dest='unjoin_agent',
                        help='Unjoin given agent(s) from given queue(s). Can be "all" to act on all agents.')

    parser.add_argument('--remove', '-r', type=str, required=False, nargs='+', dest='remove_user',
                        help='Remove given agent from given queue(s). Can be "all" to act on all agents.')

    parser.add_argument('--add', '-a', type=str, required=False, nargs='+', dest='add_user',
                        help='Add given users to given queue(s).')
    parser.add_argument('--dryrun', '-d', required=False, action='store_true',
                        help='Dry run; don\'t apply any changes')
    parser.add_argument('--token', type=str, required=False, help='admin access token to use')

    args = parser.parse_args()

    # get environment variables from .env; required for integration parameters
    load_dotenv(env_path())

    tokens = args.token or None
    if tokens is None:
        # get tokens from cache or create a new set of tokens using the integration defined in .env
        tokens = get_tokens()

    async with AsWebexSimpleApi(tokens=tokens) as api:
        # validate location parameter
        location_names = args.location or []

        # list of all locations with names matching one of the provided names
        locations = [loc for loc in await api.locations.list()
                     if not location_names or loc.name in set(location_names)]

        if not location_names:
            print(f'Considering all {len(locations)} locations')

        # set of names of matching locations
        found_location_names = set(loc.name for loc in locations)

        # Error message for each location name argument not matching an actual location
        for location_name in location_names:
            if location_name not in found_location_names:
                print(f'location "{location_name}" not found', file=sys.stderr)

        if not locations:
            print('Found no locations to work on', file=sys.stderr)
            exit(1)

        # which queues do we need to operate on?
        location_ids = set(loc.location_id for loc in locations)
        queue_names = args.queue
        all_queues = queue_names is None
        # full list of queues
        queues = await api.telephony.callqueue.list()
        # filter based on location parameter
        queues = [queue for queue in queues
                  if (all_queues or queue.name in queue_names) and queue.location_id in location_ids]

        # len of queue names for nicer output
        queue_len = max(len(queue.name) for queue in queues)

        # now we can actually go back and re-evaluate the list of locations; for the location length we only need
        # to consider locations we actually have a target queue in
        location_ids = set(queue.location_id for queue in queues)

        # max length of location names for nicely formatted output
        location_len = max(len(loc.name)
                           for loc in locations
                           if loc.location_id in location_ids)

        # get the names for join, unjoin, remove, and add
        join_agents = args.join_agent or []
        unjoin_agents = args.unjoin_agent or []
        remove_users = args.remove_user or []
        add_users = args.add_user or []

        # validate users; make sure that users exist with the provided names
        add_users = [u async for u in validate_users(user_names=add_users)]

        # apply actions to all queues
        await asyncio.gather(*[act_on_queue(queue) for queue in queues])


if __name__ == '__main__':
    # enable DEBUG logging to a file; REST log shows all requests
    logging.basicConfig(filename=os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.log'),
                        filemode='w', level=logging.DEBUG)
    asyncio.run(main())