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


if __name__ == '__main__':
    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.model_validate(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())

Using service APP tokens to access a API endpoints

The script uses service app credentials to get an access token and then use this access token to call Webex Calling APIs.

Source: service_app.py

#!/usr/bin/env python
"""
Demo for Webex service app: using service APP tokens to access a API endpoints
"""
import logging
import sys
from json import dumps, loads
from os import getenv
from os.path import basename, splitext, isfile
from typing import Optional

from dotenv import load_dotenv
from yaml import safe_load, safe_dump

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


def yml_path() -> str:
    """
    Get filename for YML file to cache access and refresh token
    """
    return f'{splitext(basename(__file__))[0]}.yml'


def env_path() -> str:
    """
    Get path to .env file to read service app settings from
    :return:
    """
    return f'{splitext(basename(__file__))[0]}.env'


def read_tokens_from_file() -> Optional[Tokens]:
    """
    Get service app tokens from cache file, return None if cache does not exist
    """
    path = yml_path()
    if not isfile(path):
        return None
    try:
        with open(path, mode='r') as f:
            data = safe_load(f)
        tokens = Tokens.model_validate(data)
    except Exception:
        return None
    return tokens


def write_tokens_to_file(tokens: Tokens):
    """
    Write tokens to cache
    """
    with open(yml_path(), mode='w') as f:
        safe_dump(tokens.model_dump(exclude_none=True), f)


def get_access_token() -> Tokens:
    """
    Get a new access token using refresh token, service app client id, service app client secret
    """
    tokens = Tokens(refresh_token=getenv('SERVICE_APP_REFRESH_TOKEN'))
    integration = Integration(client_id=getenv('SERVICE_APP_CLIENT_ID'),
                              client_secret=getenv('SERVICE_APP_CLIENT_SECRET'),
                              scopes=[], redirect_url=None)
    integration.refresh(tokens=tokens)
    write_tokens_to_file(tokens)
    return tokens


def get_tokens() -> Optional[Tokens]:
    """
    Get tokens from cache or create new access token using service app credentials
    """
    # try to read from file
    tokens = read_tokens_from_file()
    # .. or create new access token using refresh token
    if tokens is None:
        tokens = get_access_token()
    if tokens.remaining < 24 * 60 * 60:
        tokens = get_access_token()
    return tokens


def service_app():
    """
    Use service app access token to call Webex Calling API endpoints
    :return:
    """
    load_dotenv(env_path())
    # assert that all required environment variable are set
    if not all(getenv(s) for s in ('SERVICE_APP_REFRESH_TOKEN', 'SERVICE_APP_CLIENT_ID', 'SERVICE_APP_CLIENT_SECRET')):
        print(
            f'SERVICE_APP_REFRESH_TOKEN, SERVICE_APP_CLIENT_ID, and SERVICE_APP_CLIENT_SECRET need to be defined in '
            f'environment or in "{env_path()}"',
            file=sys.stderr)
        exit(1)

    # get tokens and dump to console
    tokens = get_tokens()
    print(dumps(loads(tokens.json()), indent=2))
    print()
    print('scopes:')
    print('\n'.join(f' * {s}' for s in sorted(tokens.scope.split())))

    # use tokens to access APIs
    api = WebexSimpleApi(tokens=tokens)

    users = list(api.people.list())
    print(f'{len(users)} users')

    queues = list(api.telephony.callqueue.list())
    print(f'{len(queues)} call queues')


if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    service_app()

Pool unassigned TNs on hunt groups to catch calls to unassigned TNs

This script looks for unassigned TNs and assigns them to HGs that are forwarded to the locations main number. The idea is to catch all incoming calls to unassigned TNs and handle them accordingly.

usage: catch_tns.py [-h] [–test] [–location LOCATION] [–token TOKEN]

[–cleanup]

optional arguments:

-h, –help show this help message and exit

–test test only; don’t actually apply any config

–location LOCATION Location to work on

–token TOKEN admin access token to use.

–cleanup remove all pooling HGs

Source: catch_tns.py

#!/usr/bin/env python
"""
This script looks for unassigned TNs and assigns them to HGs that are forwarded to the locations main number.
The idea is to catch all incoming calls to unassigned TNs and handle them accordingly

    usage: catch_tns.py [-h] [--test] [--location LOCATION] [--token TOKEN]
                        [--cleanup]

    optional arguments:
      -h, --help           show this help message and exit
      --test               test only; don't actually apply any config
      --location LOCATION  Location to work on
      --token TOKEN        admin access token to use.
      --cleanup            remove all pooling HGs
"""
import asyncio
import logging
import os
import sys
from argparse import ArgumentParser, Namespace
from collections import defaultdict
from operator import attrgetter
from typing import Optional, Union

from dotenv import load_dotenv

from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.common import IdAndName, AlternateNumber, RingPattern
from wxc_sdk.integration import Integration
from wxc_sdk.scopes import parse_scopes
from wxc_sdk.telephony import NumberListPhoneNumber
from wxc_sdk.telephony.forwarding import CallForwarding
from wxc_sdk.telephony.huntgroup import HuntGroup
from wxc_sdk.tokens import Tokens

POOL_HG_NAME = 'POOL_'


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 location_id_from_args(api: AsWebexSimpleApi, args: Namespace) -> Optional[str]:
    """
    Get location id for --location parameter
    """
    if not args.location:
        return None
    location = next((loc
                     for loc in await api.locations.list(name=args.location)
                     if loc.name == args.location),
                    None)
    return location and location.location_id


async def cleanup(api: AsWebexSimpleApi, args: Namespace):
    """
    clean up (delete) pooling HGs
    """
    location_id = await location_id_from_args(api, args)
    existing_hg_list = [hg for hg in await api.telephony.huntgroup.list(location_id=location_id)
                        if hg.name.startswith(POOL_HG_NAME)]

    async def delete_one(hg: HuntGroup):
        if not args.test:
            await api.telephony.huntgroup.delete_huntgroup(location_id=hg.location_id,
                                                           huntgroup_id=hg.id)
        print(f'Deleted HG "{hg.name}" in location "{hg.location_name}"')

    await asyncio.gather(*[delete_one(hg)
                           for hg in existing_hg_list])


async def pool_tns_location(api: AsWebexSimpleApi, args: Namespace, location: IdAndName,
                            tns: list[NumberListPhoneNumber]):
    """
    Pool unassigned TNs for a given location
    """

    async def add_to_hg(hg: Union[HuntGroup, str],
                        tns: list[NumberListPhoneNumber]):
        """
        Add a bunch of TNs to given HG
        :param hg: can be an existing HG or a name of a HG to be created
        :param tns: list of TNs to be added to the HG
        """
        if isinstance(hg, str):
            # create new HG
            print(f'Creating HG "{hg}" in location "{location.name}" for: '
                  f'{", ".join(tn.phone_number for tn in tns)}')
            if args.test:
                return
            settings = HuntGroup.create(name=hg,
                                        phone_number=tns[0].phone_number)
            new_id = await api.telephony.huntgroup.create(location_id=location.id,
                                                          settings=settings)
            # Get details of new HG and also the forwarding settings
            # * details are needed for the recursive call to add_to_hg .. in case we have more than one TN to add
            # * ... and forwarding settings are needed b/c we want to set CFwdAll to location's main number
            details, forwarding = await asyncio.gather(
                api.telephony.huntgroup.details(location_id=location.id, huntgroup_id=new_id),
                api.telephony.huntgroup.forwarding.settings(location_id=location.id, feature_id=new_id))
            forwarding: CallForwarding
            details: HuntGroup

            # set call forwarding
            print(f'HG "{hg}" in location "{location.name}": set CFwdAll to {main_number}')
            forwarding.always.enabled = True
            forwarding.always.destination = main_number
            await api.telephony.huntgroup.forwarding.update(location_id=location.id, feature_id=new_id,
                                                            forwarding=forwarding)
            if len(tns) > 1:
                # add the remaining as alternate numbers
                await add_to_hg(hg=details, tns=tns[1:])
            return

        # add tns as alternate numbers
        print(f'HG "{hg.name}" in location "{location.name}", adding: '
              f'{", ".join(tn.phone_number for tn in tns)}')
        if args.test:
            return
        # weirdly on GET alternate numbers are returned in alternate_number_settings ...
        alternate_numbers = hg.alternate_number_settings.alternate_numbers
        alternate_numbers.extend(AlternateNumber(phone_number=tn.phone_number,
                                                 ring_pattern=RingPattern.normal)
                                 for tn in tns)
        # ... while for an update Wx expects the alternate numbers in an alternate_number attribute
        update = HuntGroup(alternate_numbers=alternate_numbers)
        await api.telephony.huntgroup.update(location_id=location.id,
                                             huntgroup_id=hg.id,
                                             update=update)

        return

    # if the list of available TNs includes the main number then something is wonky. The main number should be owned
    # by something
    main_number = next((tn for tn in tns if tn.main_number), None)
    if main_number is not None:
        print(f'Error: main number {main_number.phone_number} in location "{location.name}" is not assigned to '
              f'anything!', file=sys.stderr)
        return

    # get main number of location
    main_number = (await api.telephony.location.details(location_id=location.id)).calling_line_id.phone_number

    # get list if "pool" HGs in location
    existing_hg_list = [hg for hg in await api.telephony.huntgroup.list(location_id=location.id)
                        if hg.name.startswith(POOL_HG_NAME)]

    # we need the details for all HGs: list() response is missing the alternate number list
    # get all HG details in parallel
    existing_hg_list = await asyncio.gather(*[api.telephony.huntgroup.details(location_id=location.id,
                                                                              huntgroup_id=hg.id)
                                              for hg in existing_hg_list])
    existing_hg_list: list[HuntGroup]

    # start with an empty list of tasks
    tasks = []

    # assign TNs to existing "pool" HGs

    # these are the HGs we can still assign some TNs to
    hgs_with_open_slots = (hg
                           for hg in existing_hg_list
                           if len(hg.alternate_number_settings.alternate_numbers) < 10)
    tns.sort(key=attrgetter('phone_number'))
    while tns:
        hg_with_open_slots = next(hgs_with_open_slots, None)
        if hg_with_open_slots is None:
            # no more HGs we can assign TNs to --> we are done here
            break

        # add some tns to this hg; hg can have max 10 alternate numbers
        tns_to_add = 10 - len(hg_with_open_slots.alternate_number_settings.alternate_numbers)
        tasks.append(add_to_hg(hg=hg_with_open_slots, tns=tns[:tns_to_add]))

        # continue w/ remaining TNs
        tns = tns[tns_to_add:]

    # assign remaining TNs to new "pool" HGs
    # .. in batches of 11
    existing_names = set(hg.name for hg in existing_hg_list)
    new_hg_names = (name
                    for i in range(1, 1000)
                    if (name := f'{POOL_HG_NAME}{i:03d}') not in existing_names)
    while tns:
        tasks.append(add_to_hg(hg=next(new_hg_names), tns=tns[:11]))
        tns = tns[11:]

    # Now run all tasks
    await asyncio.gather(*tasks)
    return


async def pool_tns(api: AsWebexSimpleApi, args: Namespace):
    """
    Assign unassigned numbers to pool HGs
    """
    location_id = await location_id_from_args(api, args)

    # get available TNs. If a location argument was present then limit to that location
    numbers = await api.telephony.phone_numbers(available=True, location_id=location_id)

    # we need to work on TNs by location ...
    tns_by_location: dict[str, list[NumberListPhoneNumber]] = defaultdict(list)
    # ... and we want to collect location information (specifically the name)
    locations: dict[str, IdAndName] = dict()
    for tn in numbers:
        locations[tn.location.id] = tn.location
        tns_by_location[tn.location.id].append(tn)

    # work on all locations in parallel
    await asyncio.gather(*[pool_tns_location(api=api, args=args,
                                             location=locations[location_id], tns=tns)
                           for location_id, tns in tns_by_location.items()])


async def catch_tns():
    """
    Main async logic
    """
    parser = ArgumentParser()
    parser.add_argument('--test', required=False, help='test only; don\'t actually apply any config',
                        action='store_true')
    parser.add_argument('--location', required=False, help='Location to work on', type=str)
    parser.add_argument('--token', type=str, required=False, help='admin access token to use.')
    parser.add_argument('--cleanup', required=False, help='remove all pooling HGs', action='store_true')
    args = parser.parse_args()

    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:
        if args.cleanup:
            await cleanup(api, args)
        else:
            await pool_tns(api, args)


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    asyncio.run(catch_tns())

Downgrade room device workspaces from Webex Calling to free calling

This script looks for workspaces in a given location (or all workspaces) and downgrades them from Webex Calling to free calling.

usage: room_devices.py [-h] [–location LOCATION] [–wsnames WSNAMES] [–test] {show,clear}

CLI tool to manage room device calling entitlements

positional arguments:

{show,clear} show: show all room devices with their calling settings, clear: remove calling

license from devices

optional arguments:

-h, –help show this help message and exit

–location LOCATION work on devices in given location

–wsnames WSNAMES file name of a file with workspace names to operate on; one name per line

–test test run only

Source: room_devices.py

#!/usr/bin/env python
"""
    usage: room_devices.py [-h] [--location LOCATION] [--wsnames WSNAMES] [--test] {show,clear}

    CLI tool to manage room device calling entitlements

    positional arguments:
      {show,clear}         show: show all room devices with their calling settings, clear: remove calling
                           license from devices

    optional arguments:
      -h, --help           show this help message and exit
      --location LOCATION  work on devices in given location
      --wsnames WSNAMES    file name of a file with workspace names to operate on; one name per line
      --test               test run only

    The tool reads environment variables from room_devices.env:
        SERVICE_APP_CLIENT_ID=<clients id of a service app created on developer.webex.com>
        SERVICE_APP_CLIENT_SECRET=<clients secret of a service app created on developer.webex.com>
        SERVICE_APP_REFRESH_TOKEN=<refresh token of the service app obtained after the service app has been
                                        authorized for an org>

    This information is used to obtain an access token required to authorize API access
                                        
    This is a super-set of the scopes the service app needs: 
        * spark-admin:workspaces_write
        * Identity:one_time_password
        * identity:placeonetimepassword_create
        * spark:people_read
        * spark-admin:workspace_locations_read
        * spark-admin:workspaces_read
        * spark:devices_write
        * spark:devices_read
        * spark:kms
        * spark-admin:devices_read
        * spark-admin:workspace_locations_write
        * spark-admin:licenses_read
        * spark-admin:telephony_config_read
        * spark-admin:telephony_config_write
        * spark-admin:devices_write
        * spark-admin:people_read

    More service app details: https://developer.webex.com/docs/service-apps

    Tokens get persisted in room_devices.yml.
"""
import asyncio
import logging
import sys
import time
from argparse import ArgumentParser
from collections import defaultdict
from functools import reduce
from itertools import chain
from os import getenv, getcwd
from os.path import splitext, basename, isfile, join
from typing import Optional

from dotenv import load_dotenv
from yaml import safe_load, safe_dump

from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.base import webex_id_to_uuid
from wxc_sdk.common import OwnerType
from wxc_sdk.devices import Device
from wxc_sdk.integration import Integration
from wxc_sdk.locations import Location
from wxc_sdk.telephony import NumberListPhoneNumber
from wxc_sdk.tokens import Tokens
from wxc_sdk.workspace_locations import WorkspaceLocation
from wxc_sdk.workspaces import Workspace, WorkspaceSupportedDevices, CallingType, WorkspaceCalling


def yml_path() -> str:
    """
    Get filename for YML file to cache access and refresh token
    """
    return f'{splitext(basename(__file__))[0]}.yml'


def env_path() -> str:
    """
    Get path to .env file to read service app settings from
    """
    return f'{splitext(basename(__file__))[0]}.env'


def read_tokens_from_file() -> Optional[Tokens]:
    """
    Get service app tokens from cache file, return None if cache does not exist or read fails
    """
    path = yml_path()
    if not isfile(path):
        return None
    try:
        with open(path, mode='r') as f:
            data = safe_load(f)
        tokens = Tokens.model_validate(data)
    except Exception:
        return None
    return tokens


def write_tokens_to_file(tokens: Tokens):
    """
    Write tokens to cache
    """
    with open(yml_path(), mode='w') as f:
        safe_dump(tokens.model_dump(exclude_none=True), f)


def get_access_token() -> Tokens:
    """
    Get a new access token using refresh token, service app client id, service app client secret
    """
    tokens = Tokens(refresh_token=getenv('SERVICE_APP_REFRESH_TOKEN'))
    integration = Integration(client_id=getenv('SERVICE_APP_CLIENT_ID'),
                              client_secret=getenv('SERVICE_APP_CLIENT_SECRET'),
                              scopes=[], redirect_url=None)
    integration.refresh(tokens=tokens)
    write_tokens_to_file(tokens)
    return tokens


def get_tokens() -> Optional[Tokens]:
    """
    Get tokens from cache or create new access token using service app credentials
    """
    # try to read from file
    tokens = read_tokens_from_file()
    # .. or create new access token using refresh token
    if tokens is None:
        tokens = get_access_token()
    if tokens.remaining < 24 * 60 * 60:
        tokens = get_access_token()
    return tokens


def main() -> int:
    """
    Main code
    """
    # parse args
    parser = ArgumentParser(prog=basename(__file__), description='CLI tool to manage room device calling entitlements')
    parser.add_argument('operation', choices=['show', 'clear'], help='show: show all room devices with their calling '
                                                                     'settings, clear: remove calling license from '
                                                                     'devices')
    parser.add_argument('--location', type=str, help='work on devices in given location')
    parser.add_argument('--wsnames', type=str, help='file name of a file with workspace names to operate on; '
                                                    'one name per line')
    parser.add_argument('--test', action='store_true', help='test run only')
    args = parser.parse_args()
    operation = args.operation
    test_run = args.test
    location = args.location
    ws_names = args.wsnames

    # get tokens; as an alternative you can just get a developer token from developer.webex.com and use:
    #   tokens = '<developer token from developer.webex.com>'
    load_dotenv(dotenv_path=env_path())
    err = ''
    tokens = None
    try:
        tokens = get_tokens()
    except Exception as e:
        err = f'{e}'
    if not tokens:
        print(f'failed to obtain access tokens: {err}', file=sys.stderr)
        return 1

    async def as_main() -> int:
        """
        Async main to be able to use concurrency
        """

        async def downgrade_workspace(ws: Workspace):
            """
            Downgrade one workspace to free calling
            """

            def log(s: str, file=sys.stdout):
                print(f'downgrade workspace "{ws.display_name:{ws_name_len}}": {s}', file=file)

            if ws.calling.type != CallingType.webex:
                raise ValueError(f'calling type is "{ws.calling.type}", not "{CallingType.webex.value}"')
            if test_run:
                log('skipping update, test run only')
            else:
                log('updating calling settings')
                update = ws.model_copy(deep=True)
                update.calling = WorkspaceCalling(type=CallingType.free)
                await api.workspaces.update(workspace_id=ws.workspace_id, settings=update)
            log('done')

        async with AsWebexSimpleApi(tokens=tokens) as api:

            # get list of locations and workspace locations
            ws_location_list, location_list = await asyncio.gather(
                api.workspace_locations.list(display_name=location),
                api.locations.list(name=location))
            location_list: list[Location]
            ws_location_list: list[WorkspaceLocation]

            # validate location argument
            if location:
                target_location = next((loc for loc in location_list if loc.name == location), None)
                target_ws_location = next((loc for loc in ws_location_list if loc.display_name == location), None)
                if not all((target_ws_location, target_location)):
                    print(f'location "{location}" not found', file=sys.stderr)
                    return 1
            else:
                target_location = None
                target_ws_location = None

            # get workspaces, numbers, and devices (in target location)
            workspaces, numbers, devices = await asyncio.gather(
                api.workspaces.list(workspace_location_id=target_ws_location and target_ws_location.id),
                api.telephony.phone_numbers(location_id=target_location and target_location.location_id,
                                            owner_type=OwnerType.place),
                api.devices.list(workspace_location_id=target_ws_location and target_ws_location.id,
                                 product_type='roomdesk')
            )
            workspaces: list[Workspace]
            numbers: list[NumberListPhoneNumber]
            devices: list[Device]

            # only workspaces supporting desk devices
            workspaces = [ws for ws in workspaces
                          if ws.supported_devices == WorkspaceSupportedDevices.collaboration_devices]

            # if a path to a file with workspace names was given, then filter based on the file contents
            if ws_names:
                with open(ws_names, mode='r') as f:
                    workspace_names = set(s_line for line in f if (s_line := line.strip()))
                workspaces = [ws for ws in workspaces
                              if ws.display_name in workspace_names]
            if not workspaces:
                print('No workspaces', file=sys.stderr)
                return 1

            # only devices in workspaces (no personal devices)
            devices = [d for d in devices if d.workspace_id is not None]

            # prepare some lookups
            workspace_locations_by_id: dict[str, WorkspaceLocation] = {wsl.id: wsl for wsl in ws_location_list}
            numbers_by_workspace_uuid: dict[str, list[NumberListPhoneNumber]] = reduce(
                lambda r, el: r[webex_id_to_uuid(el.owner.owner_id)].append(el) or r,
                numbers,
                defaultdict(list))
            devices_by_workspace_id: dict[str, list[Device]] = reduce(
                lambda r, el: r[el.workspace_id].append(el) or r,
                devices,
                defaultdict(list))

            # sort workspaces by workspace location name and workspace name; workspace location can be unset
            workspaces.sort(key=lambda ws: ('' if not ws.workspace_location_id else
                                            workspace_locations_by_id[ws.workspace_location_id].display_name,
                                            ws.display_name))
            # some field lengths for nicer output
            wsl_name_len = max(len(wsl.display_name) for wsl in ws_location_list)
            ws_name_len = max(len(ws.display_name) for ws in workspaces)

            # ... chain([1], ...) to avoid max() on empty sequence
            pn_len = max(chain([1], (len(n.phone_number) for n in numbers if n.phone_number)))
            ext_len = max(chain([1], (len(n.extension) for n in numbers if n.extension)))

            # print workspaces with workspace locations, numbers, and devices
            for workspace in workspaces:
                if not workspace.workspace_location_id:
                    wsl_name = ''
                else:
                    wsl_name = workspace_locations_by_id[workspace.workspace_location_id].display_name
                print(f'workspace location "{wsl_name:{wsl_name_len}}", '
                      f'workspace "{workspace.display_name:{ws_name_len}}"')

                # are there any numbers in that workspace?
                numbers = numbers_by_workspace_uuid.get(webex_id_to_uuid(workspace.workspace_id))
                if numbers:
                    for number in numbers:
                        print(f'  number: {number.phone_number or "-" * pn_len:{pn_len}}/'
                              f'{number.extension or "-" * ext_len:{ext_len}}')
                devices = devices_by_workspace_id.get(workspace.workspace_id)
                if devices:
                    for device in devices:
                        print(f'  device: {device.display_name}')

            if operation == 'show':
                # we are done here
                return 0

            # now we want to downgrade (disable calling) on all workspaces
            print()
            print('Starting downgrade')
            results = await asyncio.gather(*[downgrade_workspace(ws) for ws in workspaces], return_exceptions=True)

            # print errors ... if any
            for ws, result in zip(workspaces, results):
                ws: Workspace
                if isinstance(result, Exception):
                    print(f'Failed to downgrade "{ws.display_name:{ws_name_len}}": {result}', file=sys.stderr)

            if any(isinstance(r, Exception) for r in results):
                return 1
            return 0

    return asyncio.run(as_main())


if __name__ == '__main__':
    root_logger = logging.getLogger()
    h = logging.StreamHandler(stream=sys.stderr)
    h.setLevel(logging.INFO)
    root_logger.setLevel(logging.INFO)
    root_logger.addHandler(h)

    # log REST API interactions to file
    file_fmt = logging.Formatter(fmt='%(asctime)s %(levelname)s %(message)s')
    file_fmt.converter = time.gmtime

    rest_log_name = join(getcwd(), f'{splitext(basename(__file__))[0]}.log')
    rest_log_handler = logging.FileHandler(rest_log_name, mode='w')
    rest_log_handler.setLevel(logging.DEBUG)
    rest_log_handler.setFormatter(file_fmt)
    rest_logger = logging.getLogger('wxc_sdk.as_rest')
    rest_logger.setLevel(logging.DEBUG)
    rest_logger.addHandler(rest_log_handler)

    exit(main())