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(override=True)

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(override=True)


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')


if __name__ == '__main__':
    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 DeviceList
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[DeviceList]
        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(override=True)

    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(entity_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(override=True)


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])) 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(user.person_id, settings=vm_settings)
            # Read configuration after changes
            vm_settings = vm.read(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 calendarific import CalendarifiyApi, Holiday
from dotenv import load_dotenv

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

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(override=True)

    # 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 calendarific import CalendarifiyApi, Holiday
from dotenv import load_dotenv

from wxc_sdk.all_types import Event, Schedule, ScheduleType
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 = 2024
LAST_YEAR = 2026

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(override=True)

    # 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 [OPTIONS] USER_EMAIL [ON_OFF]:[on|off]

 read/update call intercept settings of a calling user.

╭─ Arguments ────────────────────────────────────────────────────────────────────────────────╮
│ *    user_email      EMAIL_TYPE         email address of user [required]                   │
│      on_off          [ON_OFF]:[on|off]  operation to apply.                                │
╰────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ Options ──────────────────────────────────────────────────────────────────────────────────╮
│ --token                     TEXT  Access token can be provided using --token argument, set │
│                                   in WEBEX_ACCESS_TOKEN environment variable or can be a   │
│                                   service app token. For the latter set environment        │
│                                   variables ('SERVICE_APP_REFRESH_TOKEN',                  │
│                                   'SERVICE_APP_CLIENT_ID', 'SERVICE_APP_CLIENT_SECRET').   │
│                                   Environment variables can also be set in service_app.env │
│ --install-completion              Install completion for the current shell.                │
│ --show-completion                 Show completion for the current shell, to copy it or     │
│                                   customize the installation.                              │
│ --help                            Show this message and exit.                              │
╰────────────────────────────────────────────────────────────────────────────────────────────╯

 Example: ./call_intercept.py bob@example.com on

Source: call_intercept.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11,<3.14"
# dependencies = [
#     "email-validator",
#     "python-dotenv",
#     "typer",
#     "wxc-sdk",
# ]
# ///
"""
Script to read/update call intercept settings of a calling user.

 Usage: call_intercept.py [OPTIONS] USER_EMAIL [ON_OFF]:[on|off]

 read/update call intercept settings of a calling user.

╭─ Arguments ────────────────────────────────────────────────────────────────────────────────╮
│ *    user_email      EMAIL_TYPE         email address of user [required]                   │
│      on_off          [ON_OFF]:[on|off]  operation to apply.                                │
╰────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ Options ──────────────────────────────────────────────────────────────────────────────────╮
│ --token                     TEXT  Access token can be provided using --token argument, set │
│                                   in WEBEX_ACCESS_TOKEN environment variable or can be a   │
│                                   service app token. For the latter set environment        │
│                                   variables ('SERVICE_APP_REFRESH_TOKEN',                  │
│                                   'SERVICE_APP_CLIENT_ID', 'SERVICE_APP_CLIENT_SECRET').   │
│                                   Environment variables can also be set in service_app.env │
│ --install-completion              Install completion for the current shell.                │
│ --show-completion                 Show completion for the current shell, to copy it or     │
│                                   customize the installation.                              │
│ --help                            Show this message and exit.                              │
╰────────────────────────────────────────────────────────────────────────────────────────────╯

 Example: ./call_intercept.py bob@example.com on
"""
import logging
import os
import sys
from enum import Enum
from typing import Optional

import typer
from dotenv import load_dotenv
from email_validator import validate_email
from service_app import SERVICE_APP_ENVS, env_path, get_tokens

from wxc_sdk import WebexSimpleApi
from wxc_sdk.person_settings.call_intercept import InterceptSetting
from wxc_sdk.rest import RestError
from wxc_sdk.tokens import Tokens

log = logging.getLogger(__name__)


def email_type(value: str):
    """
    Validate email address
    """
    validated = validate_email(value, check_deliverability=False)
    return validated.normalized


class OnOff(str, Enum):
    ON = 'on'
    OFF = 'off'


app = typer.Typer()


@app.command(name=os.path.basename(__file__),
             help='read/update call intercept settings of a calling user.',
             epilog=f'Example: {sys.argv[0]} bob@example.com on')
def main(user_email: str = typer.Argument(...,
                                          help='email address of user',
                                          parser=email_type),
         on_off: Optional[OnOff] = typer.Argument(None,
                                                  help='operation to apply.'),
         token: Optional[str] = typer.Option(None, '--token',
                                             help=f'Access token can be provided using --token argument, '
                                                  f'set in WEBEX_ACCESS_TOKEN environment variable or '
                                                  f'can be a service app token. For the latter set '
                                                  f'environment variables {SERVICE_APP_ENVS}. '
                                                  f'Environment variables can also be set in '
                                                  f'{env_path()}')):
    """
    where the magic happens
    """
    # get tokens
    load_dotenv(env_path(), override=True)

    tokens = get_tokens() if token is None else Tokens(access_token=token)
    if tokens is None:
        print(f'Access token can be provided using --token argument, set in WEBEX_ACCESS_TOKEN environment variable or '
              f'can be a service app token. For the latter set environment variables {SERVICE_APP_ENVS}. Environment '
              f'variables can '
              f'also be set in {env_path()}', file=sys.stderr)

    # 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 = 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(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 on_off:
            # action: turn on/off
            intercept = InterceptSetting.default()
            intercept.enabled = on_off == OnOff.ON
            try:
                api.person_settings.call_intercept.configure(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(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__':
    app()

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.as_api import AsWebexSimpleApi
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
from wxc_sdk.tokens import Tokens


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 f'{os.path.splitext(__file__)[0]}.env'


def yml_path() -> str:
    """
    determine path of YML file to persist tokens
    :return: path to YML file
    :rtype: str
    """
    return f'{os.path.splitext(__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(id=user.person_id)
                          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 -S uv run --script
# /// script
# requires-python = ">=3.11,<3.14"
# dependencies = [
#     "python-dotenv",
#     "wxc-sdk",
# ]
# ///
"""
Demo for Webex service app: using service APP tokens to access a API endpoints
"""

import inspect
import logging
import os
import sys
from json import dumps, loads
from os import getenv
from typing import Optional

import yaml
from dotenv import load_dotenv

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

SERVICE_APP_ENVS = ('SERVICE_APP_REFRESH_TOKEN', 'SERVICE_APP_CLIENT_ID', 'SERVICE_APP_CLIENT_SECRET')


def yml_path(*, client_id: str) -> str:
    """
    Get filename for YML file to cache access and refresh token
    """
    return f'tokens_{client_id}.yml'


def env_path() -> str:
    """
    Get path to .env file to read service app settings from
    """
    # get the file name of the calling script to determine the name of the .env file
    frame = inspect.currentframe().f_back
    file_name = inspect.getframeinfo(frame).filename
    env_name = f'{os.path.splitext(os.path.basename(file_name))[0]}.env'
    if not os.path.isfile(env_name):
        # fallback to .env specific to this file
        env_name = f'{os.path.splitext(os.path.basename(__file__))[0]}.env'
    return env_name


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


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


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


def get_tokens() -> Optional[Tokens]:
    """
    Get tokens from environment variable, cache or create new access token using service app credentials
    """
    refresh, client_id, client_secret = (os.getenv(env) for env in SERVICE_APP_ENVS)
    if not all((refresh, client_id, client_secret)):
        token = os.getenv('WEBEX_ACCESS_TOKEN')
        if token is None:
            return None
        tokens = Tokens(access_token=token)
    else:
        # try to read from file
        tokens = read_tokens_from_file(client_id=client_id)
        # .. or create new access token using refresh token
        if tokens is None:
            tokens = get_access_token(client_id=client_id, client_secret=client_secret, refresh=refresh)
        if tokens.expires_in is not None and tokens.remaining < 24 * 60 * 60:
            tokens = get_access_token(client_id=client_id, client_secret=client_secret, refresh=refresh)
    return tokens


def service_app():
    """
    Use service app access token to call Webex Calling API endpoints
    """
    load_dotenv(env_path())
    # assert that all required environment variable are set
    if not all(getenv(s) for s in SERVICE_APP_ENVS):
        print(
            f'{", ".join(SERVICE_APP_ENVS)} environment variables 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
    with WebexSimpleApi(tokens=tokens) as api:
        users = list(api.people.list())
        print(f'{len(users)} users')

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


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 AlternateNumber, IdAndName, 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(), override=True)

    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

options:
  -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
(wxc-sdk-py3.11) ➜  examples git:(master) ✗ ./room_devices.py --help
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

options:
  -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 getcwd, getenv
from os.path import basename, isfile, join, splitext
from typing import Optional

from dotenv import load_dotenv
from yaml import safe_dump, safe_load

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 CallingType, Workspace, WorkspaceCalling, WorkspaceSupportedDevices


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) 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)
                update.workspace_location_id = None
                update.location_id = None
                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) 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())

Logout users by selectively revoking authorizations

Selectively revoke authorizations.

usage: logout_users.py [-h] [--appname APPNAME] [--test] email

CLI tool to logout users by revoking user authorizations

positional arguments:
  email              single email or path to file w/ email addresses (one email address per
                     line). "all" can be used to print authorizations for all users. "all"
                     cannot be combined with other parameters and no authorizations will be
                     revoked in this case."

options:
  -h, --help         show this help message and exit
  --appname APPNAME  regular expression matching authorization application names. When missing
                     authorizations for all client ids defined in the script are revoked
  --test             test run only

Source: logout_users.py

#!/usr/bin/env python
"""
    usage: logout_users.py [-h] [--appname APPNAME] [--test] email

    CLI tool to logout users by revoking user authorizations

    positional arguments:
      email              single email or path to file w/ email addresses (one email address per line). "all" can be
                         used to print authorizations for all users. "all" cannot be combined with other parameters
                         and no authorizations will be revoked in this case."

    optional arguments:
      -h, --help         show this help message and exit
      --appname APPNAME  regular expression matching authorization application names. When missing authorizations for
                         all client ids defined in the script are revoked
      --test             test run only


    The script uses a service app to access the authorization APIs:
    https://developer.webex.com/admin/docs/api/v1/authorizations

    The service app needs these scopes: identity:tokens_write identity:tokens_read spark-admin:people_read

    Service app details are read from "logout_users.env":

        SERVICE_APP_CLIENT_ID=<client id>>
        SERVICE_APP_CLIENT_SECRET=<client secret>>
        SERVICE_APP_REFRESH_TOKEN=<service app refresh token>>
"""
import asyncio
import logging
import re
import sys
import time
from argparse import ArgumentParser
from itertools import chain
from operator import attrgetter
from os import getcwd, getenv
from os.path import join, splitext, basename, isfile
from typing import Optional

from dotenv import load_dotenv
from yaml import safe_dump, safe_load

from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.authorizations import Authorization, AuthorizationType
from wxc_sdk.integration import Integration
from wxc_sdk.people import Person
from wxc_sdk.tokens import Tokens

# list of client ids to revoke authorizations for
# add more client ids as needed
CLIENT_IDS = {
    # Webex Web Client
    'C64ab04639eefee4798f58e7bc3fe01d47161be0d97ff0d31e040a6ffe66d7f0a',
    # Webex Teams Desktop Client for Mac
    'Ccb2581f071a0714c8ab7d4777f70ebed26f1ef5f3261597f00afbb3a53c1ad88',
    # add more client ids as required, call the script with "all" parameter to identify more client ids
}


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 auth_str(auth: Authorization) -> str:
    return f'{auth.type:7}, {auth.client_id} - {auth.application_name}'


async def main() -> int:
    parser = ArgumentParser(prog=basename(__file__),
                            description='CLI tool to logout users by revoking user authorizations')
    parser.add_argument('email',
                        help='single email or path to file w/ email addresses (one email address per line). "all" can '
                             'be used to print authorizations for all users. "all" cannot be combined with other '
                             'parameters and no authorizations will be revoked in this case."')
    parser.add_argument('--appname',
                        type=str,
                        help='regular expression matching authorization application names. '
                             'When missing authorizations for all client ids defined in the '
                             'script are revoked')
    parser.add_argument('--test', action='store_true', help='test run only')
    args = parser.parse_args()
    email = args.email
    test_run = args.test
    appname = args.appname

    if appname:
        try:
            appname_re = re.compile(appname)
        except re.error as e:
            print(f'invalid regular expression for --appname: {e}')
            return 1

    async def work_on_one_email(api: AsWebexSimpleApi, user_email: str) -> int:
        """
        Work on authorizations for one email
        """
        print(f'Getting authorizations for {user_email}')
        auths = await api.authorizations.list(person_email=user_email)

        auths.sort(key=lambda a: f'{a.application_name}{a.type}')

        # determine set of authorization ids to revoke
        if appname:
            auths_to_delete = set(a.id for a in auths
                                  if appname_re.match(a.application_name))
        else:
            auths_to_delete = set(a.id for a in auths
                                  if a.client_id in CLIENT_IDS)
        if auths:
            # show all authorizations and indicate which will be revoked
            print('\n'.join(f'{user_email}: {auth_str(auth)} '
                            f'{"--> revoke" if auth.id in auths_to_delete else ""}'
                            for auth in auths))

        if not auths:
            print(f'{user_email}: no auths found')
            return 0
        if not auths_to_delete:
            print(f'{user_email}: no auths to revoke')
            return 0
        if test_run:
            print(f'{user_email}: testrun, not revoking any auths')
            return 0
        results = await asyncio.gather(*[api.authorizations.delete(authorization_id=a_id)
                                         for a_id in auths_to_delete],
                                       return_exceptions=True)
        err = False
        for r, auth_id in zip(results, auths_to_delete):
            if not isinstance(r, Exception):
                continue
            auth = next((a for a in auths if a.id == auth_id))
            if auth.type == AuthorizationType.refresh:
                # ignore errors on revoking access tokens (race condition)
                continue
            print(f'{user_email}: {auth_str(auth)}, error revoking auth: {r}')
            err = True
        if err:
            return 1
        return 0

    # 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 with AsWebexSimpleApi(tokens=tokens) as api:
        if email == 'all':
            print('Getting users...')
            users = await api.people.list()
            print(f'Getting authorizations for {len(users)} users...')
            auth_lists = await asyncio.gather(*[api.authorizations.list(person_id=user.person_id)
                                                for user in users],
                                              return_exceptions=True)
            err = False
            for user, error in zip(users, auth_lists):
                user: Person
                if isinstance(error, Exception):
                    print(f'{user.emails}: failed to get authorizations: {error}')
                    err = True
            if err:
                return 1
            auth_dict: dict[str, list[Authorization]] = {person.person_id: auth_list
                                                         for person, auth_list in zip(users, auth_lists)
                                                         if auth_list}
            print('\n'.join(chain.from_iterable((f'{user.emails[0]}: {auth_str(auth)}'
                                                 for auth in auth_dict.get(user.person_id, []))
                                                for user in sorted(users, key=attrgetter('display_name')))))

            clients = set(auth_str(auth) for auth in chain.from_iterable(auth_lists))
            print()
            print('tokens and clients:')
            print('\n'.join(sorted(clients)))
            return 0
        elif isfile(email):
            # read email addresses from file, one email address per line
            with open(email, mode='r') as f:
                emails = list(sorted(set(s_line for line in f if (s_line := line.strip()))))
        else:
            emails = [email]

        if not emails:
            print('Nothing to do')
            return 0

        # act on all emails concurrently
        results = await asyncio.gather(*[work_on_one_email(api, e)
                                         for e in emails],
                                       return_exceptions=True)
    err = next((r for r in results if isinstance(r, Exception)), None)
    if err:
        raise err
    return max(results)


if __name__ == '__main__':
    root_logger = logging.getLogger()
    h = logging.StreamHandler(stream=sys.stderr)
    h.setLevel(logging.DEBUG)
    root_logger.setLevel(logging.DEBUG)
    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(asyncio.run(main()))

Leave spaces with no activity in the last n days

Leave spaces w/o recent activity.

usage: leave_spaces.py [-h] [--days DAYS] [--token TOKEN] [--no_test] [--no_messages]
                       [--keep KEEP]

leave spaces with no activity

options:
  -h, --help            show this help message and exit
  --days DAYS, -d DAYS  days since last activity; default: 1095
  --token TOKEN         Personal access token to use. If not provided script will try to read
                        token from WEBEX_ACCESS_TOKEN environment variable.
  --no_test             Don't test; actually leave the spaces
  --no_messages         Only leave spaces that have no messages
  --keep KEEP, -k KEEP  file with list of spaces to keep

Source: leave_spaces.py

#!/usr/bin/env python
import asyncio
import logging
import os
import sys
from argparse import ArgumentParser
from collections.abc import Iterable
from datetime import datetime, timedelta
from typing import Optional

from dateutil import tz
from dotenv import load_dotenv

from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.as_rest import AsRestError
from wxc_sdk.common import RoomType
from wxc_sdk.messages import Message
from wxc_sdk.rooms import Room
from wxc_sdk.teams import Team


async def last_n_messages(api: AsWebexSimpleApi, space: Room, n: int = 10) -> list[Message]:
    """
    Get last n messages in a space
    """
    messages = []
    if not n:
        return messages
    async for message in api.messages.list_gen(room_id=space.id, max=min(n, 1000)):
        messages.append(message)
        n -= 1
        if not n:
            break
    return messages


async def latest_message_in_space(api: AsWebexSimpleApi, space: Room) -> Optional[Message]:
    """
    Get latest message in a space
    """
    last_messages = await last_n_messages(api, space, 1)
    if not last_messages:
        return None
    return last_messages[0]


async def verify_leave_space(api: AsWebexSimpleApi, space: Room, cutoff: datetime) -> bool:
    """
    Get latest message in a space and check if it's older than cutoff
    """
    latest = await latest_message_in_space(api, space)
    if not latest:
        return True
    return latest.created <= cutoff


async def leave_spaces(api: AsWebexSimpleApi, spaces: Iterable[Room]):
    """
    Leave some spaces
    """
    me = await api.people.me()
    person_id = me.person_id

    async def leave_space(space: Room):
        # get membership to delete
        try:
            memberships = await api.membership.list(room_id=space.id, person_id=person_id)
            if not memberships:
                print(f'No membership in space "{space.title}", skipping', file=sys.stderr)
                return
            # delete membership
            print(f'Leaving space "{space.title}"...')
            await api.membership.delete(memberships[0].id)
            print(f'Left space "{space.title}"')
        except AsRestError as e:
            print(f'Error leaving space "{space.title}": {e}', file=sys.stderr)
        return

    await asyncio.gather(*[leave_space(space) for space in spaces])


async def as_main():
    # parse args
    parser = ArgumentParser(description='leave spaces with no activity')
    parser.add_argument(
        '--days', '-d', type=int, required=False, default=3 * 365, help=f'days since last activity; default: {3 * 365}'
    )
    parser.add_argument(
        '--token',
        type=str,
        required=False,
        help='Personal access token to use. If not provided script will try to read token from '
        'WEBEX_ACCESS_TOKEN environment variable.',
    )
    parser.add_argument('--no_test', action='store_true', required=False, help="Don't test; actually leave the spaces")
    parser.add_argument(
        '--no_messages', action='store_true', required=False, help='Only leave spaces that have no messages'
    )
    parser.add_argument('--keep', '-k', type=str, required=False, help='file with list of spaces to keep')
    args = parser.parse_args()

    load_dotenv(override=True)
    token = args.token or os.getenv('WEBEX_ACCESS_TOKEN')
    if not token:
        print('No token provided and WEBEX_ACCESS_TOKEN not set in environment', file=sys.stderr)
        exit(1)
    cutoff = datetime.now(tz=tz.UTC) - timedelta(days=args.days)
    if args.keep:
        try:
            with open(args.keep) as f:
                keep = set(l.strip() for l in f if l)
        except FileNotFoundError:
            print(f'file "{args.keep}" not found', file=sys.stderr)
            exit(1)
    else:
        keep = set()
    async with AsWebexSimpleApi(tokens=token, concurrent_requests=100) as api:
        # check token
        try:
            await api.people.me()
        except AsRestError as e:
            if e.status == 401:
                print(f'Token seems to be invalid: {e}', file=sys.stderr)
                exit(1)
            raise

        print('Listing spaces and teams...')
        spaces, teams_list = await asyncio.gather(api.rooms.list(max=1000), api.teams.list())
        # we can't leave direct spaces anyway; so ignore them from the start
        spaces = [space for space in spaces if space.type != RoomType.direct]
        print(f'Found {len(spaces)} spaces')
        print(f'Found {len(teams_list)} teams')
        teams = {team.id: team for team in teams_list}

        # identify spaces to leave based on last activity
        leave = [space for space in spaces if space.last_activity and space.last_activity <= cutoff]

        # sometimes the last_activity information seems to be "off" try to get the latest message for each space to
        # verify latest activity
        print(f'Getting latest message for {len(leave)} spaces...')
        latest_messages = await asyncio.gather(*[latest_message_in_space(api, space) for space in leave])
        validated_leave: list[tuple[Room, Optional[datetime]]] = []
        for space, latest_message in zip(leave, latest_messages):
            space: Room
            latest_message: Message
            # don't leave general spaces of Teams
            # A general space is a space that has the same name as the Team it belongs to
            team: Team
            if space.team_id and (team := teams.get(space.team_id, None)) and space.title == team.name:
                print(f'Don\'t leave general space "{space.title}" of Team "{team.name}"')
                continue

            # don't leave spaces in keep file
            if space.title.strip() in keep:
                print(f'Don\'t leave space "{space.title}", found space name in keep file')
                continue

            # if only spaces with no messages should be considered and we have a message in the space then don't leave
            if args.no_messages and latest_message is not None:
                print(
                    f'Space "{space.title}" has messages, latest is {latest_message.created:%Y.%m.%d %H:%M:%S}, '
                    f'last activity is {space.last_activity:%Y.%m.%d %H:%M:%S} - not leaving'
                )
                continue

            # if no latest message or latest message is older than cutoff, consider leaving
            if latest_message is None or latest_message.created <= cutoff:
                latest_messages: Message
                validated_leave.append((space, latest_message and latest_message.created))
                continue
            print(
                f'Latest message in "{space.title}" is {latest_message.created:%Y.%m.%d %H:%M:%S}, '
                f'last activity is {space.last_activity:%Y.%m.%d %H:%M:%S} - not leaving'
            )
        print()
        print(f'Found {len(validated_leave)} spaces to leave:')

        # sort spaces by latest activity or latest message
        validated_leave.sort(key=lambda x: max(x[0].last_activity, x[1]) if x[1] else x[0].last_activity, reverse=False)

        for space, latest in validated_leave:
            print(f'Leave "{space.title}"')
            print(
                f'   last activity {space.last_activity:%Y.%m.%d %H:%M:%S}, latest message '
                f'{f"{latest:%Y.%m.%d %H:%M:%S}" if latest else "none"}'
            )

        if args.no_test:
            # actually try to leave the spaces
            await leave_spaces(api, (space for space, _ in validated_leave))


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

Provision location level access codes from a CSV file

Provision location level access codes from a CSV file

 Usage: access_codes.py [OPTIONS] CSV_FILE

 Provision location level access codes from a CSV file

╭─ Arguments ────────────────────────────────────────────────────────────────────────────────╮
│ *    csv_file      PATH  CSV file with access codes [required]                             │
╰────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ Options ──────────────────────────────────────────────────────────────────────────────────╮
│ --dry-run                         Do not make any changes                                  │
│ --token                     TEXT  Access token can be provided using --token argument, set │
│                                   in WEBEX_ACCESS_TOKEN environment variable or can be a   │
│                                   service app token. For the latter set environment        │
│                                   variables ('SERVICE_APP_REFRESH_TOKEN',                  │
│                                   'SERVICE_APP_CLIENT_ID', 'SERVICE_APP_CLIENT_SECRET').   │
│                                   Environment variables can also be set in                 │
│                                   service_app.env                                          │
│ --install-completion              Install completion for the current shell.                │
│ --show-completion                 Show completion for the current shell, to copy it or     │
│                                   customize the installation.                              │
│ --help                            Show this message and exit.                              │
╰────────────────────────────────────────────────────────────────────────────────────────────╯

Source: access_codes.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11,<3.14"
# dependencies = [
#     "typer",
#     "wxc-sdk",
#     "python-dotenv",
# ]
# ///
"""
Provision location level access codes from a CSV file
the CSV file should have the following columns:
- location: location name
- code: the access code
- description: a description of the access code
- operation: one of 'add', 'delete'

Usage: access_codes.py [OPTIONS] CSV_FILE

 Provision location level access codes from a CSV file

╭─ Arguments ────────────────────────────────────────────────────────────────────────────────╮
│ *    csv_file      PATH  CSV file with access codes [required]                             │
╰────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ Options ──────────────────────────────────────────────────────────────────────────────────╮
│ --dry-run                         Do not make any changes                                  │
│ --token                     TEXT  Access token can be provided using --token argument, set │
│                                   in WEBEX_ACCESS_TOKEN environment variable or can be a   │
│                                   service app token. For the latter set environment        │
│                                   variables ('SERVICE_APP_REFRESH_TOKEN',                  │
│                                   'SERVICE_APP_CLIENT_ID', 'SERVICE_APP_CLIENT_SECRET').   │
│                                   Environment variables can also be set in                 │
│                                   access_codes.env                                         │
│ --install-completion              Install completion for the current shell.                │
│ --show-completion                 Show completion for the current shell, to copy it or     │
│                                   customize the installation.                              │
│ --help                            Show this message and exit.                              │
╰────────────────────────────────────────────────────────────────────────────────────────────╯

"""
import asyncio
import csv
import logging
import sys
from collections import defaultdict
from functools import reduce
from os.path import basename
from pathlib import Path
from typing import Literal, Optional

import typer
from dotenv import load_dotenv
from pydantic import BaseModel, TypeAdapter
from service_app import SERVICE_APP_ENVS, env_path, get_tokens

from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.common import AuthCode
from wxc_sdk.telephony.location import TelephonyLocation
from wxc_sdk.tokens import Tokens


class CSVOperation(BaseModel):
    """
    row in CSV file: Operation to be performed on an access code
    """
    location: str
    code: str
    description: str
    operation: Literal['add', 'delete']


async def process_operations(*, operations: list[CSVOperation], token: str, dry_run:bool):
    """
    Process the operations from the CSV file in parallel
    """

    # noinspection PyShadowingNames
    async def process_one_location(*, location: str, operations: list[CSVOperation]):
        """
        process operations for one location
        """
        # find the location
        loc = locations.get(location)
        if loc is None:
            print(f'Location {location} not found', sys.stderr)
            return

        # get existing codes in location
        ac_codes: dict[str, AuthCode] = {ac.code: ac
                                         for ac in await api.telephony.access_codes.read(location_id=loc.location_id)}

        tasks = []
        delete_operations = [operation
                             for operation in operations
                             if operation.operation == 'delete']
        non_existing = [op
                        for op in delete_operations
                        if op.code not in ac_codes]
        if non_existing:
            print(f'Location {location}: access codes not found: {", ".join(op.code for op in non_existing)}',
                  file=sys.stderr)
        to_delete = [ac
                     for ac in delete_operations
                     if ac.code in ac_codes]
        print(f'Location {location}: deleting access codes: {", ".join(op.code for op in to_delete)}')
        if to_delete:
            tasks.append(api.telephony.access_codes.delete_codes(location_id=loc.location_id,
                                                                 access_codes=[op.code for op in to_delete]))

        add_operations = [operation
                          for operation in operations
                          if operation.operation == 'add']
        existing = [op for op in add_operations if op.code in ac_codes]
        if existing:
            print(f'Location {location}: access codes already exist: {", ".join(op.code for op in existing)}',
                  file=sys.stderr)
        to_add = [ac for ac in add_operations if ac.code not in ac_codes]
        print(f'Location {location}: adding access codes: {", ".join(op.code for op in to_add)}')
        if to_add:
            tasks.append(api.telephony.access_codes.create(location_id=loc.location_id,
                                                           access_codes=[AuthCode(code=op.code,
                                                                                  description=op.description)
                                                                         for op in to_add]))
        if tasks and not dry_run:
            await asyncio.gather(*tasks)

    async with AsWebexSimpleApi(tokens=token) as api:
        # get locations
        locations: dict[str, TelephonyLocation] = {loc.name: loc for loc in await api.telephony.locations.list()}

        # group operations by location
        operations_by_location: dict[str, list[CSVOperation]] = reduce(
            lambda acc, op: acc[op.location].append(op) or acc,
            operations,
            defaultdict(list))

        # process all locations in parallel
        await asyncio.gather(*[process_one_location(location=loc, operations=ops)
                               for loc, ops in operations_by_location.items()])
    return


app = typer.Typer()


@app.command(name=basename(__file__),
             help='Provision location level access codes from a CSV file')
def main(csv_file: Path = typer.Argument(exists=True,
                                         help='CSV file with access codes'),
         dry_run: bool = typer.Option(False, '--dry-run',
                                      help='Do not make any changes'),

         token: Optional[str] = typer.Option(None, '--token',
                                             help=f'Access token can be provided using --token argument, '
                                                  f'set in WEBEX_ACCESS_TOKEN environment variable or '
                                                  f'can be a service app token. For the latter set '
                                                  f'environment variables {SERVICE_APP_ENVS}. '
                                                  f'Environment variables can also be set in '
                                                  f'{env_path()}')):
    # get access token
    load_dotenv(env_path(), override=True)
    tokens = get_tokens() if token is None else Tokens(access_token=token)
    if tokens is None:
        print(f'Access token can be provided using --token argument, set in WEBEX_ACCESS_TOKEN environment variable or '
              f'can be a service app token. For the latter set environment variables {SERVICE_APP_ENVS}. Environment '
              f'variables can '
              f'also be set in {env_path()}', file=sys.stderr)
        exit(1)

    csv_file = str(csv_file)
    # read CSV file
    with open(csv_file) as f:
        reader = csv.DictReader(f)
        data = list(reader)

    # validate the data from the CSV file
    csv_operations = TypeAdapter(list[CSVOperation]).validate_python(data)

    asyncio.run(process_operations(operations=csv_operations,
                                   token=tokens,
                                   dry_run=dry_run))


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

Bulk assign/unassign agents to/from call queues

Bulk assign/unassign agents to/from call queues

usage: queue_agents.py [-h] (--add | --remove) --queues QUEUES --agent AGENT [--token TOKEN]
                       [--debug] [--har] [--dry-run]

Bulk manage agents in call queues

options:
  -h, --help       show this help message and exit
  --add            Add agent(s) to specified queues
  --remove         Remove agent(s) from specified queues
  --queues QUEUES  Text file with list of call queue names (one per line). Each line should
                   be the a location name and a queue name separated by a colon. Example:
                   "Location1:Queue1"
  --agent AGENT    Single agent email address or text file with agent email addresses (one
                   per line)
  --token TOKEN    admin access token to use. If no token is given then the script will try
                   to use service app tokens. The service app parameters are read from
                   environment variables SERVICE_APP_ID, SERvICE_APP_SECRET, and
                   SERVICE_APP_REFRESH. These parameters can also be defined in
                   "queue_agents.env" file. Service app tokens are cached in
                   "queue_agents.yml". If no access token is passed and no service app is
                   defined then the script falls back to try to read an access token from
                   environment variable WEBEX_ACCESS_TOKEN.
  --debug          Enable debug output
  --har            Enable HAR output
  --dry-run        Simulate the operation without making actual changes

Example: queue_agents.py --queues queues.txt --agent agents.txt --add

Source: queue_agents.py

#!/usr/bin/env python3
"""
Bulk manage agents in call queues

usage: queue_agents.py [-h] (--add | --remove) --queues QUEUES --agent AGENT [--token TOKEN]

Bulk manage agents in call queues

options:
  -h, --help       show this help message and exit
  --add            Add agent(s) to specified queues
  --remove         Remove agent(s) from specified queues
  --queues QUEUES  Text file with list of call queue names (one per line). Each line should be the a
                   location name and a queue name separated by a colon. Example: "Location1:Queue1"
  --agent AGENT    Single agent email address or text file with agent email addresses (one per line)
  --token TOKEN    admin access token to use. If no token is given then the script will try to use
                   service app tokens. The service app parameters are read from environment variables
                   SERVICE_APP_ID, SERvICE_APP_SECRET, and SERVICE_APP_REFRESH. These parameters can
                   also be defined in "queue_agents.env" file. Service app tokens are cached in
                   "queue_agents.yml". If no access token is passed and no service app is defined then
                   the script falls back to try to read an access token from environment variable
                   WEBEX_ACCESS_TOKEN.

Example: queue_agents.py --queues queues.txt --agent agents.txt --add
"""

import argparse
import asyncio
import logging
import os
import sys
from collections.abc import Iterable
from contextlib import contextmanager
from typing import List, Optional

import yaml
from dotenv import load_dotenv

from wxc_sdk import Tokens
from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.har_writer import HarWriter
from wxc_sdk.integration import Integration
from wxc_sdk.people import Person
from wxc_sdk.telephony.callqueue import CallQueue
from wxc_sdk.telephony.hg_and_cq import Agent


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


def env_path() -> str:
    """
    Get path to .env file to read service app settings from
    :return:
    """
    return f'{os.path.splitext(os.path.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 os.path.isfile(path):
        return None
    try:
        with open(path) as f:
            data = yaml.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:
        yaml.safe_dump(tokens.model_dump(exclude_none=True), f)


def get_access_token() -> Optional[Tokens]:
    """
    Get a new access token using refresh token, service app client id, service app client secret
    """
    env_vars = ('SERVICE_APP_ID', 'SERVICE_APP_SECRET', 'SERVICE_APP_REFRESH')
    app_id, app_secret, app_refresh = (os.getenv(var) for var in env_vars)
    if not all((app_id, app_secret, app_refresh)):
        return None
    tokens = Tokens(refresh_token=app_refresh)
    integration = Integration(client_id=app_id, client_secret=app_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 is None:
            return None
    if tokens.remaining < 24 * 60 * 60:
        tokens = get_access_token()
    return tokens


def read_file_lines(filename: str) -> List[str]:
    """
    Read lines from a file, stripping whitespace and removing empty lines.

    Args:
        filename (str): Path to the input file

    Returns:
        List[str]: Cleaned list of lines from the file

    Raises:
        FileNotFoundError: If the specified file does not exist
        PermissionError: If there are permission issues reading the file
    """
    try:
        with open(filename) as f:
            return [line.strip() for line in f if line.strip()]
    except FileNotFoundError:
        print(f"Error: File '{filename}' not found.", file=sys.stderr)
        sys.exit(1)
    except PermissionError:
        print(f"Error: Permission denied reading file '{filename}'.", file=sys.stderr)
        sys.exit(1)


async def process_one_queue(
    *, api: AsWebexSimpleApi, queue: CallQueue, agents: list[Person], action: str, dry_run: bool = True
):
    """
    Process adding or removing agents from a single call queue.
    """
    # get agents
    details = await api.telephony.callqueue.details(location_id=queue.location_id, queue_id=queue.id)

    agent: Agent
    agents_in_queue = set(agent.agent_id for agent in details.agents)
    agent_ids = set(person.person_id for person in agents)
    if action == 'add':
        agents_to_add = agent_ids - agents_in_queue
        if not agents_to_add:
            print(f'All agents are already in queue {queue.name} in location {queue.location_name}. Skipping.')
            return
        agent_str = ', '.join(
            sorted(
                f'{person.display_name}({person.emails[0]})' for person in agents if person.person_id in agents_to_add
            )
        )
        print(f'Adding agents to queue {queue.name} in location {queue.location_name}: {agent_str}')
        details.agents.extend([Agent(agent_id=agent_id) for agent_id in agents_to_add])
    else:
        agents_to_remove = agents_in_queue & agent_ids
        if not agents_to_remove:
            print(f'No agents to remove from queue {queue.name} in location {queue.location_name}. Skipping.')
            return
        agent_str = ', '.join(
            sorted(
                f'{person.display_name}({person.emails[0]})'
                for person in agents
                if person.person_id in agents_to_remove
            )
        )
        print(f'Removing agents from queue {queue.name} in location {queue.location_name}: {agent_str}')
        details.agents = [agent for agent in details.agents if agent.agent_id not in agents_to_remove]
    update = CallQueue(agents=details.agents)
    if not dry_run:
        await api.telephony.callqueue.update(location_id=queue.location_id, queue_id=queue.id, update=update)


async def validate_queues(api: AsWebexSimpleApi, queues: Iterable[str]) -> list[CallQueue]:
    """
    Validate queue names and return a list of CallQueue objects
    """
    # validate queue names
    existing_queues = {f'{queue.location_name}:{queue.name}': queue for queue in await api.telephony.callqueue.list()}
    validated_queues = []
    for queue in queues:
        if queue not in existing_queues:
            print(f"Queue '{queue}' does not exist. Skipping.", file=sys.stderr)
        else:
            validated_queues.append(existing_queues[queue])

    return validated_queues


async def validate_users(api: AsWebexSimpleApi, users: Iterable[str]) -> list[Person]:
    """
    Validate user emails and return a list of Person objects
    """
    existing_users = {user.emails[0].lower(): user for user in await api.people.list()}
    validated_users = []
    for user in users:
        if user.lower() not in existing_users:
            print(f"User '{user}' does not exist. Skipping.", file=sys.stderr)
        else:
            validated_users.append(existing_users[user.lower()])

    return validated_users


def main():
    """
    Main CLI script entry point for managing call queue agents.
    """
    parser = argparse.ArgumentParser(
        description='Bulk manage agents in call queues',
        epilog='Example: %(prog)s --queues queues.txt --agent agents.txt --add',
    )

    # Mutually exclusive group for add/remove actions
    action_group = parser.add_mutually_exclusive_group(required=True)
    action_group.add_argument(
        '--add', action='store_const', const='add', dest='action', help='Add agent(s) to specified queues'
    )
    action_group.add_argument(
        '--remove', action='store_const', const='remove', dest='action', help='Remove agent(s) from specified queues'
    )

    # Input source arguments
    parser.add_argument(
        '--queues',
        required=True,
        help='Text file with list of call queue names (one per line). Each line should be the a '
        'location name and a queue name separated by a colon. Example: "Location1:Queue1"',
    )
    parser.add_argument(
        '--agent',
        required=True,
        help='Single agent email address or text file with agent email addresses (one per line)',
    )
    parser.add_argument(
        '--token',
        type=str,
        required=False,
        help=f'admin access token to use. If no token is given then the script will try to use '
        f'service app tokens. The service app parameters are read from environment variables '
        f'SERVICE_APP_ID, SERvICE_APP_SECRET, and SERVICE_APP_REFRESH. These parameters can also '
        f'be defined in "{os.path.splitext(os.path.basename(__file__))[0]}.env" file. Service '
        f'app tokens are cached in "'
        f'{os.path.splitext(os.path.basename(__file__))[0]}.yml". If no access token is passed '
        f'and no service app '
        f'is defined then the script falls back to try to read an access token from environment '
        f'variable WEBEX_ACCESS_TOKEN.',
    )

    # Debug and HAR output arguments
    parser.add_argument('--debug', action='store_true', help='Enable debug output')
    parser.add_argument('--har', action='store_true', help='Enable HAR output')

    # fry run option
    parser.add_argument('--dry-run', action='store_true', help='Simulate the operation without making actual changes')

    # Parse arguments
    args = parser.parse_args()

    # Get access token
    token = args.token
    load_dotenv(os.path.join(os.getcwd(), f'{os.path.splitext(os.path.basename(__file__))[0]}.env'))

    token = token or (tokens := get_tokens()) and tokens.access_token

    # Read queues from file
    queues = read_file_lines(args.queues)

    # Determine if agent is a file or a single agent email
    try:
        agents = read_file_lines(args.agent) if os.path.isfile(args.agent) else [args.agent]
    except FileNotFoundError:
        agents = [args.agent]

    if args.debug:
        logging.basicConfig(level=logging.DEBUG)

    async def as_main():
        """
        Async main entry point
        """

        @contextmanager
        def har_writer():
            """
            optional context manager to write HAR file
            """
            if args.har:
                with HarWriter(f'{os.path.splitext(os.path.basename(__file__))[0]}.har', api):
                    yield None
            else:
                yield None

        async with AsWebexSimpleApi(tokens=token) as api:
            with har_writer():
                # validate queues and agents
                validated_queues, validated_agents = await asyncio.gather(
                    validate_queues(api, queues), validate_users(api, agents)
                )
                if not validated_queues:
                    print('No valid queues found. Exiting.', file=sys.stderr)
                    sys.exit(1)
                if not validated_agents:
                    print('No valid agents found. Exiting.', file=sys.stderr)
                    sys.exit(1)
                validated_queues: List[CallQueue]
                validated_agents: List[Person]

                # process all queues in parallel
                await asyncio.gather(
                    *[
                        process_one_queue(
                            api=api, queue=queue, agents=validated_agents, action=args.action, dry_run=args.dry_run
                        )
                        for queue in validated_queues
                    ],
                    return_exceptions=False,
                )
            # with
        return

    asyncio.run(as_main())

    print(f'Completed {args.action} operation for {len(agents)} agent(s) across {len(queues)} queue(s).')


if __name__ == '__main__':
    main()

Bulk add phone numbers to locations

Bulk add TNs to Webex Calling locations

 Usage: add_numbers.py [OPTIONS] FILE

 Add TNs to Webex Calling locations

╭─ Arguments ────────────────────────────────────────────────────────────────────────────────╮
│ *    file      PATH  CSV file with location names and TNs [required]                       │
╰────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ Options ──────────────────────────────────────────────────────────────────────────────────╮
│ --dry-run                         Do not make any changes                                  │
│ --verbose                         Print debug information                                  │
│ --log-file                  PATH  Log file. If extension is .har, log in HAR format        │
│ --token                     TEXT  Access token can be provided using --token argument, set │
│                                   in WEBEX_ACCESS_TOKEN environment variable or can be a   │
│                                   service app token. For the latter set environment        │
│                                   variables ('SERVICE_APP_REFRESH_TOKEN',                  │
│                                   'SERVICE_APP_CLIENT_ID', 'SERVICE_APP_CLIENT_SECRET').   │
│                                   Environment variables can also be set in add_numbers.env │
│ --inactive                        Add TNs as inactive                                      │
│ --install-completion              Install completion for the current shell.                │
│ --show-completion                 Show completion for the current shell, to copy it or     │
│                                   customize the installation.                              │
│ --help                            Show this message and exit.                              │
╰────────────────────────────────────────────────────────────────────────────────────────────╯

 Example: ./add_numbers.py add_numbers.csv --log-file add_numbers.har --dry-run

Source: add_numbers.py

#!/usr/bin/env -S uv run --script
#
# /// script
# requires-python = ">=3.11,<3.14"
# dependencies = [
#     "python-dotenv",
#     "typer",
#     "wxc-sdk",
# ]
# ///
"""
Add TNs to a locations.

 Usage: add_numbers.py [OPTIONS] FILE

 Add TNs to Webex Calling locations

╭─ Arguments ────────────────────────────────────────────────────────────────────────────────╮
│ *    file      PATH  CSV file with location names and TNs [required]                       │
╰────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ Options ──────────────────────────────────────────────────────────────────────────────────╮
│ --dry-run                         Do not make any changes                                  │
│ --verbose                         Print debug information                                  │
│ --log-file                  PATH  Log file. If extension is .har, log in HAR format        │
│ --token                     TEXT  Access token can be provided using --token argument, set │
│                                   in WEBEX_ACCESS_TOKEN environment variable or can be a   │
│                                   service app token. For the latter set environment        │
│                                   variables ('SERVICE_APP_REFRESH_TOKEN',                  │
│                                   'SERVICE_APP_CLIENT_ID', 'SERVICE_APP_CLIENT_SECRET').   │
│                                   Environment variables can also be set in add_numbers.env │
│ --inactive                        Add TNs as inactive                                      │
│ --install-completion              Install completion for the current shell.                │
│ --show-completion                 Show completion for the current shell, to copy it or     │
│                                   customize the installation.                              │
│ --help                            Show this message and exit.                              │
╰────────────────────────────────────────────────────────────────────────────────────────────╯

 Example: ./add_numbers.py add_numbers.csv --log-file add_numbers.har --dry-run

"""

import asyncio
import csv
import logging
import os
import sys
from collections import Counter, defaultdict
from contextlib import contextmanager
from functools import wraps
from itertools import chain
from pathlib import Path
from typing import Optional

import typer
from dotenv import load_dotenv
from service_app import SERVICE_APP_ENVS, env_path, get_tokens

from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.common import NumberState
from wxc_sdk.har_writer import HarWriter
from wxc_sdk.tokens import Tokens

BATCH_SIZE = 10


@contextmanager
def setup_logging(*, api: AsWebexSimpleApi, verbose: bool, log_file: Optional[Path]):
    """
    Set up logging
    """

    @contextmanager
    def file_handler():
        if not log_file:
            yield
        else:
            # log to file or to HAR
            log_file_str = str(log_file)
            if os.path.splitext(log_file_str)[-1].lower() == '.har':
                with HarWriter(api=api, path=log_file_str):
                    yield
            else:
                f_handler = logging.FileHandler(log_file_str)
                f_handler.setLevel(logging.DEBUG)
                f_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
                logging.getLogger().addHandler(f_handler)
                yield
        return

    logging.getLogger().setLevel(logging.DEBUG)
    # create a console logging handler
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG if verbose else logging.INFO)
    # console_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
    logging.getLogger().addHandler(console_handler)
    with file_handler():
        yield


def read_csv(file: str) -> dict[str, list[str]]:
    """
    Read CSV file with location names and TNs
    """
    if not os.path.isfile(file):
        logging.error(f'File {file} does not exist')
        exit(1)
    err = False
    locations_and_tns: dict[str, list[str]] = defaultdict(list)
    with open(file) as f:
        reader = csv.reader(f)
        for row in reader:
            if not row:
                continue
            if len(row) != 2:
                logging.error(f'Invalid row: {row}')
                err = True
                continue
            location, tn = row
            logging.debug(f'Location: {location}, TN: {tn}')
            locations_and_tns[location].append(tn)
    if err:
        logging.error('Errors in the CSV file')
        exit(1)
    return locations_and_tns


async def verify_location(api: AsWebexSimpleApi, location_name: str) -> Optional[str]:
    """
    Verify that a location exists and return location id
    """
    try:
        location = next(
            (l for l in await api.telephony.locations.list(name=location_name) if l.name == location_name), None
        )
        return location and location.location_id
    except Exception as e:
        logging.error(f'Failed to get location {location_name}: {e}')
        return None


async def validate_tns(api: AsWebexSimpleApi, tns: list[str]) -> bool:
    """
    Validate TNs and return validity
    """
    tn_counter = Counter(tns)
    err = False
    for tn, count in tn_counter.items():
        if count > 1:
            err = True
            logging.error(f'TN {tn} is duplicated')
    if err:
        return False

    # validate numbers in batches
    try:
        validations = await asyncio.gather(
            *[api.telephony.validate_phone_numbers(tns[i : i + BATCH_SIZE]) for i in range(0, len(tns), BATCH_SIZE)]
        )
    except Exception as e:
        logging.error(f'Failed to validate TNs: {e}')
        return False

    ok_tns = []
    err = False
    for status in chain.from_iterable(v.phone_numbers for v in validations):
        if status.ok:
            ok_tns.append(status.phone_number)
        else:
            err = True
            logging.error(f'TN {status.phone_number}: {status.state} ')
    return not err


async def add_tns(api: AsWebexSimpleApi, location_id: str, tns: list[str], inactive: bool):
    """
    Add TNs to a location
    """
    number_state = NumberState.inactive if inactive else NumberState.active
    try:
        # add TNs in batches
        await asyncio.gather(
            *[
                api.telephony.location.number.add(
                    location_id=location_id, phone_numbers=tns[i : i + BATCH_SIZE], state=number_state
                )
                for i in range(0, len(tns), BATCH_SIZE)
            ]
        )
    except Exception as e:
        logging.error(f'Failed to add TNs: {e}')
        return False
    return True


def async_command(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        return asyncio.run(f(*args, **kwargs))

    return wrapper


app = typer.Typer()


@app.command(
    epilog=f'Example: {sys.argv[0]} add_numbers.csv --log-file add_numbers.har --dry-run',
    help='Add TNs to Webex Calling locations',
)
@async_command
async def add_numbers(
    file: Path = typer.Argument(exists=True, help='CSV file with location names and TNs'),
    dry_run: bool = typer.Option(False, '--dry-run', help='Do not make any changes'),
    verbose: bool = typer.Option(False, '--verbose', help='Print debug information'),
    log_file: Optional[Path] = typer.Option(
        None, '--log-file', help='Log file. If extension is .har, log in HAR format'
    ),
    token: Optional[str] = typer.Option(
        None,
        '--token',
        help=f'Access token can be provided using --token argument, '
        f'set in WEBEX_ACCESS_TOKEN environment variable or '
        f'can be a service app token. For the latter set '
        f'environment variables {SERVICE_APP_ENVS}. '
        f'Environment variables can also be set in '
        f'{env_path()}',
    ),
    inactive: bool = typer.Option(False, '--inactive', help='Add TNs as inactive'),
):
    """
    Add TNs to Webex Calling locations
    """
    load_dotenv(env_path(), override=True)
    tokens = get_tokens() if token is None else Tokens(access_token=token)
    if tokens is None:
        print(
            f'Access token can be provided using --token argument, set in WEBEX_ACCESS_TOKEN environment variable or '
            f'can be a service app token. For the latter set environment variables {SERVICE_APP_ENVS}. Environment '
            f'variables can '
            f'also be set in {env_path()}',
            file=sys.stderr,
        )
        exit(1)

    async with AsWebexSimpleApi(tokens=tokens) as api:
        with setup_logging(api=api, verbose=verbose, log_file=log_file):
            # validate the access token
            try:
                await api.people.me()
            except Exception as e:
                logging.error(f'Failed to get identity: {e}')
                logging.error('Token might be invalid')
                exit(1)
            # read location names and TNs from a CSV file
            logging.info(f'Reading file {file}')
            locations_and_tns = read_csv(file)

            # validate the location names
            logging.info('Validating location names...')
            location_ids = await asyncio.gather(
                *[verify_location(api, location) for location in locations_and_tns.keys()]
            )
            if not all(location_ids):
                for location_name, location_id in zip(locations_and_tns.keys(), location_ids):
                    if not location_id:
                        logging.error(f'Location {location_name} does not exist')
                exit(1)

            # validate the TNs
            logging.info('Validating TNs...')
            tn_list = list(chain.from_iterable(locations_and_tns.values()))
            validation = await validate_tns(api, tn_list)
            if not validation:
                exit(1)

            # add TNs to the locations
            if not dry_run:
                logging.info('Adding TNs...')
                results = await asyncio.gather(
                    *[
                        add_tns(api, location_id, tns, inactive)
                        for location_id, tns in zip(location_ids, locations_and_tns.values())
                    ]
                )
                if not all(results):
                    exit(1)
            #
            logging.info('Done')
        # end of logging context
    # end of API context
    return


if __name__ == '__main__':
    app()

Bulk add outgoing call permission patterns to locations

Bulk add outgoing call permission patterns to locations

usage: ocp_pattern.py [-h] [--token TOKEN] [--dry-run] [--verbose] [--log-file LOG_FILE]
                      location patterns

Provision OCP patterns for one or all locations

positional arguments:
  location             Location to provision OCP patterns for. Use "all" to provision for
                       all locations
  patterns             File with patterns to provision. File has one pattern per line. Use
                       "remove" to remove all patterns previously provisioned by the
                       script

optional arguments:
  -h, --help           show this help message and exit
  --token TOKEN        Access token can be provided using --token argument, set in
                       WEBEX_ACCESS_TOKEN environment variable or can be a service app
                       token. For the latter set environment variables
                       ('SERVICE_APP_REFRESH_TOKEN', 'SERVICE_APP_CLIENT_ID',
                       'SERVICE_APP_CLIENT_SECRET'). Environment variables can also be set
                       in ocp_pattern.env
  --dry-run            Dry run, do not provision anything
  --verbose            Print debug information
  --log-file LOG_FILE  Log file. If extension is .har, log in HAR format

Example: ocp_pattern.py all ocp_pattern.txt --log-file ocp_pattern.har --dry-run

Source: ocp_pattern.py

#!/usr/bin/env python3
"""
Provision OCP patterns for one or all locations

usage: ocp_pattern.py [-h] [--token TOKEN] [--dry-run] [--verbose] [--log-file LOG_FILE]
                      location patterns

Provision OCP patterns for one or all locations

positional arguments:
  location             Location to provision OCP patterns for. Use "all" to provision for
                       all locations
  patterns             File with patterns to provision. File has one pattern per line. Use
                       "remove" to remove all patterns previously provisioned by the
                       script

optional arguments:
  -h, --help           show this help message and exit
  --token TOKEN        Access token can be provided using --token argument, set in
                       WEBEX_ACCESS_TOKEN environment variable or can be a service app
                       token. For the latter set environment variables
                       ('SERVICE_APP_REFRESH_TOKEN', 'SERVICE_APP_CLIENT_ID',
                       'SERVICE_APP_CLIENT_SECRET'). Environment variables can also be set
                       in ocp_pattern.env
  --dry-run            Dry run, do not provision anything
  --verbose            Print debug information
  --log-file LOG_FILE  Log file. If extension is .har, log in HAR format

Example: ocp_pattern.py all ocp_pattern.txt --log-file ocp_pattern.har --dry-run
"""

import argparse
import asyncio
import logging
import os
import sys
from contextlib import contextmanager

from dotenv import load_dotenv

from examples.service_app import SERVICE_APP_ENVS, env_path, get_tokens
from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.as_rest import AsRestError
from wxc_sdk.har_writer import HarWriter
from wxc_sdk.person_settings.permissions_out import Action, DigitPattern
from wxc_sdk.telephony.location import TelephonyLocation
from wxc_sdk.tokens import Tokens


async def work_on_one_location(
    api: AsWebexSimpleApi, location: TelephonyLocation, pattern_list: list[str], dry_run: bool
):
    """
    Work on one location, create, update or remove patterns
    """

    async def create(pattern: str):
        try:
            await dapi.create(
                location.location_id,
                DigitPattern(pattern=pattern, name=f'ocp-{pattern}', action=Action.allow, transfer_enabled=True),
            )
            print(f'{location.name}: created pattern {pattern}')
        except AsRestError as e:
            print(f'{location.name}: failed to create pattern {pattern}, error: {e}')
            raise

    async def remove(pattern: DigitPattern):
        try:
            await dapi.delete(location.location_id, pattern.id)
            print(f'{location.name}: removed pattern {pattern.pattern}')
        except AsRestError as e:
            print(f'{location.name}: failed to remove pattern {pattern.pattern}, error: {e}')
            raise

    async def update(pattern: DigitPattern):
        try:
            await dapi.update(
                location.location_id,
                DigitPattern(
                    pattern=pattern.pattern, name=f'ocp-{pattern.pattern}', action=Action.allow, transfer_enabled=True
                ),
            )
            print(f'{location.name}: updated pattern {pattern.pattern}')
        except AsRestError as e:
            print(f'{location.name}: failed to update pattern {pattern.pattern}, error: {e}')
            raise

    pattern_set = set(pattern_list)
    dapi = api.telephony.permissions_out.digit_patterns

    # get ocp patterns for location
    location_digit_patterns = await dapi.get_digit_patterns(location.location_id)

    existing_patterns = [
        pattern for pattern in location_digit_patterns.digit_patterns if pattern.pattern in pattern_set
    ]
    missing_patterns = [
        pattern
        for pattern in pattern_list
        if pattern not in set(map(lambda p: p.pattern, location_digit_patterns.digit_patterns))
    ]
    to_be_removed = [
        pattern
        for pattern in location_digit_patterns.digit_patterns
        if pattern.name.startswith('ocp-') and pattern.pattern not in pattern_set
    ]
    tasks = []

    # remove patterns
    for remove_pattern in to_be_removed:
        # remove pattern
        print(f'{location.name}: remove pattern {remove_pattern.pattern}')
        if not dry_run:
            tasks.append(remove(remove_pattern))

    # add missing patterns
    for pattern in missing_patterns:
        # add pattern
        print(f'{location.name}: add pattern {pattern}')
        if not dry_run:
            tasks.append(create(pattern))

    # check existing patterns and update if action is not "allow"
    for existing in existing_patterns:
        if existing.action == Action.allow and existing.transfer_enabled:
            # pattern already exists and is allowed
            continue
        # update existing pattern
        print(f'{location.name}: update pattern {existing.pattern}')
        if not dry_run:
            tasks.append(update(existing))
    if tasks:
        # run tasks
        await asyncio.gather(*tasks)
    return


@contextmanager
def setup_logging(args: argparse.Namespace, api: AsWebexSimpleApi):
    """
    Set up logging
    """

    @contextmanager
    def file_handler(log_file: str):
        if not log_file:
            yield
        else:
            # log to file or to HAR
            if os.path.splitext(log_file)[-1].lower() == '.har':
                with HarWriter(api=api, path=log_file):
                    yield
            else:
                f_handler = logging.FileHandler(args.log_file)
                f_handler.setLevel(logging.DEBUG)
                f_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
                logging.getLogger().addHandler(f_handler)
                yield
        return

    logging.getLogger().setLevel(logging.DEBUG)
    # create a console logging handler
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG if args.verbose else logging.INFO)
    console_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
    logging.getLogger().addHandler(console_handler)
    with file_handler(args.log_file):
        yield


async def main():
    parser = argparse.ArgumentParser(
        description='Provision OCP patterns for one or all locations',
        epilog='Example: %(prog)s all ocp_pattern.txt --log-file ocp_pattern.har --dry-run',
    )
    parser.add_argument(
        'location', type=str, help='Location to provision OCP patterns for. Use "all" to provision for all locations'
    )
    parser.add_argument(
        'patterns',
        type=str,
        help='File with patterns to provision. File has one pattern '
        'per line. Use "remove" to remove all patterns previously '
        'provisioned '
        'by the script',
    )
    parser.add_argument(
        '--token',
        help=f'Access token can be provided using --token argument, set in '
        f'WEBEX_ACCESS_TOKEN environment variable or can be a service app token. For '
        f'the latter set environment variables {SERVICE_APP_ENVS}. Environment '
        f'variables can also be set in {env_path()}',
    )
    parser.add_argument('--dry-run', action='store_true', help='Dry run, do not provision anything')
    parser.add_argument('--verbose', action='store_true', help='Print debug information')
    parser.add_argument('--log-file', help='Log file. If extension is .har, log in HAR format')

    args = parser.parse_args()
    location_name = args.location
    pattern_file = args.patterns
    dry_run = args.dry_run

    # get tokens
    # if token is provided use that token, else try to read from file
    load_dotenv(env_path(), override=True)
    tokens = get_tokens() if args.token is None else Tokens(access_token=args.token)
    if tokens is None:
        print(
            f'Access token can be provided using --token argument, set in WEBEX_ACCESS_TOKEN environment variable '
            f'or '
            f'can be a service app token. For the latter set environment variables {SERVICE_APP_ENVS}. Environment '
            f'variables can '
            f'also be set in {env_path()}',
            file=sys.stderr,
        )
        exit(1)
    async with AsWebexSimpleApi(tokens=tokens, concurrent_requests=100) as api:
        with setup_logging(args, api):
            # validate location
            if location_name.lower() == 'all':
                # all locations
                location_list = await api.telephony.locations.list()
                location_list.sort(key=lambda loc: loc.name)
            else:
                # single location
                location_list = [
                    loc for loc in await api.telephony.locations.list(name=location_name) if loc.name == location_name
                ]
                if not location_list:
                    print(f'Location {location_name} not found', file=sys.stderr)
                    exit(1)

            # read patterns from given file
            if pattern_file.lower() == 'remove':
                # remove all patterns
                pattern_list = []
            else:
                try:
                    with open(pattern_file) as f:
                        pattern_list = [ps for p in f.readlines() if (ps := p.strip()) and not p.startswith('#')]
                except FileNotFoundError:
                    print(f'File {pattern_file} not found', file=sys.stderr)
                    exit(1)

            # apply changes to all locations
            print(f'Working on {len(location_list)} location(s), {len(pattern_list)} patterns')
            await asyncio.gather(*[work_on_one_location(api, loc, pattern_list, dry_run) for loc in location_list])
    return


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

Bulk provisioning of 3rd party devices in Workspaces

Bulk provisioning of 3rd party devices in Workspaces

usage: workspace_w_3rd_party.py [-h] [--token TOKEN] [--dry-run]
                                [--log-file LOG_FILE] [--cleanup]
                                csv [output]

Provision workspaces with 3rd party devices.

positional arguments:
  csv                  CSV with workspaces to provision. CSV has the
                       following columns: * workspace name: the workspace
                       will be created * location name: must be an
                       existing location * extension (optional); if
                       missing a new extension will be generated starting
                       at 2000 * MAC address; if empty a new (dummy) MAC
                       address will be generated as DEAD-DEAD-XXXX *
                       password (optional); if missing a new (random
                       password will be generated
  output               Output CSV with the provisioning results. Not
                       required in dry-run mode

optional arguments:
  -h, --help           show this help message and exit
  --token TOKEN        Access token can be provided using --token
                       argument, set in WEBEX_ACCESS_TOKEN environment
                       variable or can be a service app token. For the
                       latter set environment variables
                       ('SERVICE_APP_REFRESH_TOKEN',
                       'SERVICE_APP_CLIENT_ID',
                       'SERVICE_APP_CLIENT_SECRET'). Environment variables
                       can also be set in workspace_w_3rd_party.env
  --dry-run            Dry run, do not provision anything
  --log-file LOG_FILE  Log file. If extension is .har, log in HAR format
  --cleanup            remove workspaces

Example: workspace_w_3rd_party.py input.csv output.csv --log-file log.har

Source: workspace_w_3rd_party.py

#!/usr/bin/env python
"""
Create workspaces with 3rd party devices.
From a CSV read:
    * workspace name: if workspace doesn't exist a new workspace will be created
    * location name: must be an existing location
    * extension (optional); if missing a new extension will be generated starting at 2000
    * MAC address; if empty a new (dummy) MAC address will be generated as DEAD-DEAD-XXXX
    * password (optional); if missing a new (random password will be generated
The output is a CSV file with the following columns:
    * workspace name
    * location name
    * extension
    * MAC address
    * password
    * outbound proxy
    * SIP user name
    * line/port

usage: workspace_w_3rd_party.py [-h] [--token TOKEN] [--dry-run]
                                [--log-file LOG_FILE] [--cleanup]
                                csv [output]

Provision workspaces with 3rd party devices.

positional arguments:
  csv                  CSV with workspaces to provision. CSV has the
                       following columns: * workspace name: the workspace
                       will be created * location name: must be an
                       existing location * extension (optional); if
                       missing a new extension will be generated starting
                       at 2000 * MAC address; if empty a new (dummy) MAC
                       address will be generated as DEAD-DEAD-XXXX *
                       password (optional); if missing a new (random
                       password will be generated
  output               Output CSV with the provisioning results. Not
                       required in dry-run mode

optional arguments:
  -h, --help           show this help message and exit
  --token TOKEN        Access token can be provided using --token
                       argument, set in WEBEX_ACCESS_TOKEN environment
                       variable or can be a service app token. For the
                       latter set environment variables
                       ('SERVICE_APP_REFRESH_TOKEN',
                       'SERVICE_APP_CLIENT_ID',
                       'SERVICE_APP_CLIENT_SECRET'). Environment variables
                       can also be set in workspace_w_3rd_party.env
  --dry-run            Dry run, do not provision anything
  --log-file LOG_FILE  Log file. If extension is .har, log in HAR format
  --cleanup            remove workspaces

Example: workspace_w_3rd_party.py input.csv output.csv --log-file log.har
"""

import argparse
import asyncio
import csv
import logging
import os
import sys
from collections import defaultdict
from collections.abc import Generator
from contextlib import contextmanager
from dataclasses import dataclass, field
from itertools import chain, zip_longest

from dotenv import load_dotenv

from examples.service_app import SERVICE_APP_ENVS, env_path, get_tokens
from open_api.generated.Shared.workspaces_auto import WorkspaceType
from wxc_sdk.as_api import AsWebexSimpleApi
from wxc_sdk.common import DevicePlatform
from wxc_sdk.har_writer import HarWriter
from wxc_sdk.licenses import License
from wxc_sdk.telephony.devices import MACState, MACValidationResponse
from wxc_sdk.telephony.location import TelephonyLocation
from wxc_sdk.tokens import Tokens
from wxc_sdk.workspaces import (
    CallingType,
    Workspace,
    WorkspaceCalling,
    WorkspaceSupportedDevices,
    WorkspaceWebexCalling,
)

MAC_VALIDATION_BATCH_SIZE = 100


@dataclass
class CSVRow:
    """
    CSV row with workspace and location name
    """

    workspace_name: str
    location_name: str
    extension: str
    mac_address: str
    password: str
    workspace: Workspace = field(default=None, init=False)
    calling_license_id: str = field(default=None, init=False)
    location: TelephonyLocation = field(default=None, init=False)
    outbound_proxy: str = field(default=None, init=False)
    sip_user_name: str = field(default=None, init=False)
    line_port: str = field(default=None, init=False)

    def __post_init__(self):
        # clean up some data
        self.workspace_name = self.workspace_name.strip()
        self.location_name = self.location_name.strip()
        self.extension = self.extension.strip() or None
        self.mac_address = self.mac_address.strip().lower() or None
        self.password = self.password.strip() or None

    @classmethod
    def from_csv(cls, csv_path: str) -> Generator['CSVRow', None, None]:
        """
        Yield CSVRow instances from CSV file
        """
        err = False
        with open(csv_path, newline='') as csv_file:
            reader = csv.reader(csv_file)
            for row_number, row in enumerate(reader, 1):
                try:
                    yield cls(*row)
                except TypeError as te:
                    err = True
                    print(f'Failed to parse row {row_number}: {te}', file=sys.stderr)
                    continue
        if err:
            print(f'Failed to parse {csv_path}', file=sys.stderr)
            exit(1)
        return


@contextmanager
def setup_logging(args: argparse.Namespace, api: AsWebexSimpleApi):
    """
    Set up logging
    """

    @contextmanager
    def file_handler(log_file: str):
        if not log_file:
            yield
        else:
            # log to file or to HAR
            if os.path.splitext(log_file)[-1].lower() == '.har':
                with HarWriter(api=api, path=log_file):
                    yield
            else:
                f_handler = logging.FileHandler(args.log_file)
                f_handler.setLevel(logging.DEBUG)
                f_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
                logging.getLogger().addHandler(f_handler)
                yield
        return

    logging.getLogger().setLevel(logging.DEBUG)
    # create a console logging handler
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
    logging.getLogger().addHandler(console_handler)
    with file_handler(args.log_file):
        yield


# list of validation errors
ValidationResult = list[tuple[int, str]]


async def validate_extensions(
    *, api: AsWebexSimpleApi, location: TelephonyLocation, indexed_rows: list[tuple[int, CSVRow]]
) -> ValidationResult:
    """
    Make sure that extensions are unique; if no extension is provided, generate a new one
    """
    extensions_in_location = {
        number.extension
        for number in await api.telephony.phone_numbers(location_id=location.location_id)
        if number.extension is not None
    }
    errors = []
    for row_index, csv_row in indexed_rows:
        if csv_row.extension:
            if csv_row.extension in extensions_in_location:
                errors.append((row_index, f'Duplicate extension "{csv_row.extension}" in location "{location.name}"'))
        else:
            # generate new extension
            csv_row.extension = next(str(ext) for ext in range(2000, 10000) if str(ext) not in extensions_in_location)
            print(f'Row {row_index}: Generated new extension "{csv_row.extension}"')
        extensions_in_location.add(csv_row.extension)

    return errors


async def generate_passwords(
    *, api: AsWebexSimpleApi, location: TelephonyLocation, indexed_rows: list[tuple[int, CSVRow]]
) -> ValidationResult:
    """
    Generate random passwords for rows without a password
    """
    new_passwords = await asyncio.gather(
        *[
            api.telephony.location.generate_password(location_id=location.location_id)
            for _, csv_row in indexed_rows
            if not csv_row.password
        ]
    )

    for row_index, csv_row in indexed_rows:
        if not csv_row.password:
            # generate new password
            csv_row.password = new_passwords.pop(0)
            print(f'Row {row_index}: Generated new password "{csv_row.password}"')
    return []


async def validate_locations_and_extensions(
    *, api: AsWebexSimpleApi, csv_rows: list[CSVRow], cleanup: bool
) -> ValidationResult:
    """
    Locations must exist and extensions must be unique
    """
    if cleanup:
        return []
    locations = {location.name: location for location in await api.telephony.locations.list()}
    errors = []
    for row_index, csv_row in enumerate(csv_rows, 1):
        location = locations.get(csv_row.location_name)
        if not location:
            errors.append((row_index, f'Location "{csv_row.location_name}" does not exist'))
        csv_row.location = location

    # for all existing locations validate extensions
    csv_rows_by_location: dict[str, list[tuple[int, CSVRow]]] = defaultdict(list)
    for row_index, csv_row in enumerate(csv_rows, 1):
        if csv_row.location_name in locations:
            csv_rows_by_location[csv_row.location_name].append((row_index, csv_row))
    tasks = [
        validate_extensions(api=api, location=locations[l_name], indexed_rows=indexed_rows)
        for l_name, indexed_rows in csv_rows_by_location.items()
    ]
    # also generate passwords for rows without a password
    tasks.extend(
        generate_passwords(api=api, location=locations[l_name], indexed_rows=indexed_rows)
        for l_name, indexed_rows in csv_rows_by_location.items()
    )
    location_results = await asyncio.gather(*tasks)
    errors.extend(chain.from_iterable(location_results))
    return errors


async def assign_new_mac_addresses(*, api: AsWebexSimpleApi, csv_rows: list[CSVRow]) -> ValidationResult:
    """
    Get new MAC addresses for rows where no MAC address is provided
    """
    number_of_missing_mac_addresses = sum(1 for csv_row in csv_rows if not csv_row.mac_address)

    if not number_of_missing_mac_addresses:
        return []
    mac_prefix = 'DEADDEAD'
    mac_addresses_in_csv = {csv_row.mac_address for csv_row in csv_rows if csv_row.mac_address}

    def mac_candidates() -> Generator[str, None, None]:
        for v in range(0, 65536):
            candidate = f'{mac_prefix}{hex(v)[2:].zfill(4).upper()}'
            if candidate in mac_addresses_in_csv:
                # skip mac addresses that are already in the csv
                continue
            yield f'{mac_prefix}{hex(v)[2:].zfill(4).upper()}'

    new_mac_addresses = []

    # test macs in batches
    batch_args = [mac_candidates()] * MAC_VALIDATION_BATCH_SIZE

    # noinspection PyArgumentList
    batches = zip_longest(*batch_args)
    for batch in batches:
        validation_result = await api.telephony.devices.validate_macs(macs=list(batch))
        errored_macs = set(ms.mac for ms in (validation_result.mac_status or []) if ms.state != MACState.available)
        new_mac_addresses.extend(mac for mac in batch if mac not in errored_macs)
        if len(new_mac_addresses) >= number_of_missing_mac_addresses:
            break
    for row_index, csv_row in enumerate(csv_rows, 1):
        if not csv_row.mac_address:
            # generate new MAC address
            csv_row.mac_address = new_mac_addresses.pop(0)
            print(f'Row {row_index}: Generated new MAC address "{csv_row.mac_address}"')
    return []


async def mac_addresses_available(*, api: AsWebexSimpleApi, csv_rows: list[CSVRow]) -> ValidationResult:
    """
    Check if MAC addresses provided in CSV are available
    """
    errors = []
    mac_addresses = list(set(csv_row.mac_address for csv_row in csv_rows if csv_row.mac_address))
    if not mac_addresses:
        return []

    # validate in batches
    batches = [
        mac_addresses[i : i + MAC_VALIDATION_BATCH_SIZE]
        for i in range(0, len(mac_addresses), MAC_VALIDATION_BATCH_SIZE)
    ]
    results = await asyncio.gather(*[api.telephony.devices.validate_macs(macs=batch) for batch in batches])
    results: list[MACValidationResponse]

    errored_macs: dict[str, str] = dict()
    for result in results:
        errored_macs.update(
            (ms.mac, f'{ms.state}, {ms.message}') for ms in (result.mac_status or []) if ms.state != MACState.available
        )

    if not errored_macs:
        return []
    for row_index, csv_row in enumerate(csv_rows, 1):
        if csv_row.mac_address and csv_row.mac_address in errored_macs:
            errors.append((row_index, f'MAC address "{csv_row.mac_address}": {errored_macs[csv_row.mac_address]}'))
    return errors


async def validate_mac_addresses(*, api: AsWebexSimpleApi, csv_rows: list[CSVRow], cleanup: bool) -> ValidationResult:
    """
    mac addresses must be unique and available
    """
    if cleanup:
        return []
    mac_addresses = set()
    errors = []
    # check if provided MAC addresses are unique
    for row_index, csv_row in enumerate(csv_rows, 1):
        if csv_row.mac_address:
            if csv_row.mac_address in mac_addresses:
                errors.append((row_index, f'Duplicate MAC address "{csv_row.mac_address}"'))
            mac_addresses.add(csv_row.mac_address)

    results = await asyncio.gather(
        assign_new_mac_addresses(api=api, csv_rows=csv_rows), mac_addresses_available(api=api, csv_rows=csv_rows)
    )
    errors.extend(chain.from_iterable(results))
    return errors


async def validate_workspaces_and_licenses(
    *, api: AsWebexSimpleApi, csv_rows: list[CSVRow], cleanup: bool
) -> ValidationResult:
    """
    Workspaces should not exist, also get license for new workspaces
    """
    tasks = [api.workspaces.list()]
    if not cleanup:
        tasks.append(api.licenses.list())
    results = await asyncio.gather(*tasks)
    workspace_list = results[0]
    if cleanup:
        licenses = []
    else:
        licenses = results[1]
    workspace_list: list[Workspace]
    licenses: list[License]
    workspaces = {ws.display_name: ws for ws in workspace_list}

    def calling_license_id() -> Generator[str, None, None]:
        """
        calling license id for license with available entitlement
        """
        candidate_licenses = [lic for lic in licenses if lic.webex_calling_workspaces or lic.webex_calling_professional]
        # make sure to consume workspace licenses first
        candidate_licenses.sort(key=lambda x: x.name, reverse=True)
        for lic in candidate_licenses:
            while lic.consumed_units < lic.total_units:
                lic.consumed_units += 1
                yield lic.license_id
        return

    errors = []
    license_id_gen = calling_license_id()
    for row_index, csv_row in enumerate(csv_rows, 1):
        # check if workspace exists
        ws = workspaces.get(csv_row.workspace_name)
        csv_row.workspace = ws
        if cleanup:
            if not ws:
                errors.append((row_index, f'Workspace "{csv_row.workspace_name}" does not exist'))
            continue
        if ws:
            errors.append((row_index, f'Workspace "{csv_row.workspace_name}" already exists'))
        # get a license for the workspace
        try:
            license_id = next(license_id_gen)
        except StopIteration:
            errors.append((row_index, f'No more licenses available for "{csv_row.workspace_name}"'))
            continue
        csv_row.calling_license_id = license_id
    return errors


async def validate_and_prepare(*, api: AsWebexSimpleApi, csv_rows: list[CSVRow], cleanup: bool) -> None:
    """
    validate csv and prepare for provisioning
        * location exists
        * extensions are unique (if provided)
        * MAC addresses are unique (if provided)
        * workspace names are unique
        * no devices in the workspace
    """
    results = await asyncio.gather(
        validate_locations_and_extensions(api=api, csv_rows=csv_rows, cleanup=cleanup),
        validate_mac_addresses(api=api, csv_rows=csv_rows, cleanup=cleanup),
        validate_workspaces_and_licenses(api=api, csv_rows=csv_rows, cleanup=cleanup),
    )
    errors = list(chain.from_iterable(results))
    errors: ValidationResult
    errors.sort(key=lambda x: x[0])
    if errors:
        print('Validation errors:', file=sys.stderr)
        for row_index, error in errors:
            print(f'Row {row_index}: {error}', file=sys.stderr)
        exit(1)
    return


async def delete_workspaces(*, api: AsWebexSimpleApi, csv_rows: list[CSVRow], dry_run: bool) -> None:
    """
    cleanup: delete workspaces
    """

    async def delete_one_workspace(workspace: Workspace) -> None:
        if dry_run:
            print(f'Delete workspace "{workspace.display_name}"')
        else:
            await api.workspaces.delete_workspace(workspace_id=workspace.workspace_id)
            print(f'Deleted workspace "{workspace.display_name}"')
        return

    await asyncio.gather(*[delete_one_workspace(csv_row.workspace) for csv_row in csv_rows if csv_row.workspace])


async def provision_row(*, api: AsWebexSimpleApi, csv_row: CSVRow) -> None:
    """
    Provision a single row
    """
    # create workspace
    settings = Workspace(
        location_id=csv_row.location.location_id,
        display_name=csv_row.workspace_name,
        type=WorkspaceType.desk,
        capacity=1,
        supported_devices=WorkspaceSupportedDevices.phones,
        device_platform=DevicePlatform.cisco,
        calling=WorkspaceCalling(
            type=CallingType.webex,
            webex_calling=WorkspaceWebexCalling(
                licenses=[csv_row.calling_license_id],
                extension=csv_row.extension,
                location_id=csv_row.location.location_id,
            ),
        ),
    )
    workspace = await api.workspaces.create(settings=settings)
    print(f'Provisioned workspace "{workspace.display_name}"')

    # create device
    device = await api.devices.create_by_mac_address(
        mac=csv_row.mac_address,
        workspace_id=workspace.workspace_id,
        model='Generic IPPhone Customer Managed',
        password=csv_row.password,
    )

    details = await api.telephony.devices.details(device_id=device.device_id)
    print(f'Provisioned device in workspace "{workspace.display_name}"')
    csv_row.sip_user_name = details.owner.sip_user_name
    csv_row.line_port = details.owner.line_port
    csv_row.outbound_proxy = details.proxy.outbound_proxy

    return


def main():
    async def as_main():
        # read CSV file
        csv_rows = list(CSVRow.from_csv(csv_file))
        async with AsWebexSimpleApi(tokens=tokens) as api:
            with setup_logging(args, api):
                # validation and preparation for provisioning
                await validate_and_prepare(api=api, csv_rows=csv_rows, cleanup=args.cleanup)
                if args.cleanup:
                    await delete_workspaces(api=api, csv_rows=csv_rows, dry_run=args.dry_run)
                    return

                if args.dry_run:
                    print('Dry run, not provisioning anything')
                    return
                await asyncio.gather(*[provision_row(api=api, csv_row=csv_row) for csv_row in csv_rows])
                # write output
                if args.output:
                    with open(args.output, 'w', newline='') as output:
                        writer = csv.writer(output)
                        writer.writerow(
                            [
                                'workspace_name',
                                'location_name',
                                'extension',
                                'mac_address',
                                'password',
                                'outbound_proxy',
                                'sip_user_name',
                                'line_port',
                            ]
                        )
                        for csv_row in csv_rows:
                            writer.writerow(
                                [
                                    csv_row.workspace_name,
                                    csv_row.location_name,
                                    csv_row.extension,
                                    csv_row.mac_address,
                                    csv_row.password,
                                    csv_row.outbound_proxy,
                                    csv_row.sip_user_name,
                                    csv_row.line_port,
                                ]
                            )
                    # for
                # with open
            # with setup_logging
        # async with AsWebexSimpleApi
        return

    # parse arguments
    parser = argparse.ArgumentParser(
        description='Provision workspaces with 3rd party devices.',
        epilog='Example: %(prog)s input.csv output.csv --log-file log.har',
    )
    parser.add_argument(
        'csv',
        type=str,
        help="""CSV with workspaces to provision. CSV has the following columns:
    * workspace name: the workspace will be created
    * location name: must be an existing location
    * extension (optional); if missing a new extension will be generated starting at 2000
    * MAC address; if empty a new (dummy) MAC address will be generated as DEAD-DEAD-XXXX
    * password (optional); if missing a new (random password will be generated""",
    )
    parser.add_argument(
        'output', nargs='?', type=str, help='Output CSV with the provisioning results. Not required in dry-run mode'
    )
    parser.add_argument(
        '--token',
        help=f'Access token can be provided using --token argument, set in '
        f'WEBEX_ACCESS_TOKEN environment variable or can be a service app token. For '
        f'the latter set environment variables {SERVICE_APP_ENVS}. Environment '
        f'variables can also be set in {env_path()}',
    )
    parser.add_argument('--dry-run', action='store_true', help='Dry run, do not provision anything')
    parser.add_argument('--log-file', help='Log file. If extension is .har, log in HAR format')
    parser.add_argument('--cleanup', action='store_true', help='remove workspaces')
    args = parser.parse_args()
    if not any((args.output, args.dry_run, args.cleanup)):
        parser.error('Output file is required if not dry-run cleanup mode')
    if args.output and (args.dry_run or args.cleanup):
        parser.error('Output file is not used in dry-run cleanup mode')
    csv_file = args.csv
    if not os.path.isfile(csv_file):
        print(f'File {csv_file} does not exist', file=sys.stderr)
        exit(1)

    # read tokens
    load_dotenv(env_path())
    tokens = get_tokens() if args.token is None else Tokens(access_token=args.token)
    if tokens is None:
        print(
            f'Access token can be provided using --token argument, set in WEBEX_ACCESS_TOKEN environment variable '
            f'or can be a service app token. For the latter set environment variables {SERVICE_APP_ENVS}. Environment '
            f'variables can also be set in {env_path()}',
            file=sys.stderr,
        )
        exit(1)

    asyncio.run(as_main())


if __name__ == '__main__':
    main()