Unity Connection Provisioning (CUPI)
via Provisioning Portal API

For the native REST APIs, this lab will start with Cisco Unity Connection's Provisioning API (CUPI). You have already configured the generic REST portion, and we have extended that class for you with the CUPI-specific components, namely the header and then the response parsing. Specifically, in flaskr/cuc/v1/cupi.py, a CUPI class has been built which extends the REST class using a base_url of /vmrest and a header defined as:

headers = {
    'Accept': 'application/json',
    'Content-Type': 'application/json'
}

In this CUPI class, we have two methods:

  • _cupi_request() - sends the request using the REST _send_request(), checks for basic errors and then returns the parsed response from _cupi_parse_response(). The _cupi_request() method takes the following variables:
    • api_method: The API method, such as "users" that will be used with the existing base_url to form a complete url, such as "/vmrest/users"
    • parameters: A dictionary of URL parameters to be sent, such as {'offset': 10}
    • payload: The payload to be sent, used with a POST or PUT
    • http_method: The request verb. CUPI only supports 'GET', 'PUT', 'POST', and 'DELETE'
  • _cupi_parse_response() - this function parses the response and then performs the following tasks:
    1. Decodes the JSON response and looks at its contents to make sure they are returned consistently (i.e. handle the case for a single item)
    2. If the results content is just a string, then return it. This is the situation you observed in the response of a POST message.
    3. In some situations, Unity Connection will return a web page describing an error exception. Instead of the whole web page, only the Exception message will be returned.

With the CUPI class built, you can initialize this class. This will be done globally, so that the same session can be used and reused by subsequent requests. With the class initialized, you can implement functions that send requests similar to what you did with Postman, such as a Get Version. When that is successful, you can implement business functions, searching for users to import, importing a user, modifying a user, and deleting a user account. This should be done using just the user ID, so the user import function, for instance, will need to first look up a user to get the pkid, to reuse in the import. Each of these functions will be tested using your Swagger UI page.

In this section, you will implement the following Unity Connection Provisioning API tasks:

  1. Get Version
  2. Get LDAP Users Available for Import
  3. Import an LDAP User by User Id
  4. Get a User by ID
  5. Modify user settings and voicemail PIN by User Id
  6. Delete the voicemail account by User Id
  7. Verification

Step 1 - Get Version

As with Postman, start with something simple: retrieving the Unity Connection version. If you recall, we sent a GET request to /version/product/, which is what you can recreate here.

  1. In your VS Code tab, open up flaskr/api/v1/cuc.py
  2. Near the top, you can instantiate the CUPI object. This will allow you to re-use the same object--and more importantly, re-use the connections--for each request.

    from flask import request
    from flask_restx import Namespace, Resource
    from flaskr.cuc.v1.cupi import CUPI
    from flaskr.api.v1.parsers import cuc_importldap_user_post_args
    from flaskr.api.v1.parsers import cuc_users_get_args
    from flaskr.api.v1.parsers import cuc_users_put_args
    from os import getenv
    
    api = Namespace('cuc', description='Cisco Unity Connection APIs')
    
    myCUPI = CUPI(getenv('CUC_HOSTNAME'), getenv('CUC_USERNAME'), getenv('CUC_PASSWORD'), port=getenv('CUC_PORT'))

  3. In the cuc_version_api() class, under the get() function, replace the pass line with the following highlighted line. Since the CUPI object has already been instantiated as myCUPI, you can simply call the _cupi_request() method. Remember that the base_url for this object is /vmrest, so the full URL requested will be /vmrest/version/product/ using the default http_method of GET.

    @api.route("/version")
    class cuc_version_api(Resource):
        def get(self):
            """
            Retrieves Unity Connection version.
            """
            return myCUPI._cupi_request("version/product/")

  4. Save this file.

Step 2 - Get LDAP Users Available for Import

Now that you are able to perform CUPI queries, build the front-end API to implement the LDAP users query, to find out who is available to be imported into the system.

  1. In your VS Code tab, open up flaskr/api/v1/cuc.py
  2. In the cuc_import_ldapuser_api() class, add the following highlighted line after the arguments are read. The get_search_params() function uses the argments to create a valid filter, like query=(alias is user1). Feel free to take a look at get_search_params() in this same file. Then you can call the _cupi_request() method using the parameters built from the arguments.

    @api.route("/ldap_users")
    class cuc_import_ldapuser_api(Resource):
        @api.expect(cuc_users_get_args, validate=True)
        def get(self):
            """
            Retrieves LDAP users synched to Unity Connection.
            """
            # Read arguments: column, match_type, search, sortorder, rowsPerPage, pageNumber
            args = cuc_users_get_args.parse_args(request)
            params = get_search_params(args)
    
            return myCUPI._cupi_request("import/users/ldap", parameters=params)

  3. Save this file.

When you were working with Postman and, for example, importing users into Unity Connection, you would also retrieve a list of users and then look for the pkid, which you would later use in the POST request to import that specific user. You don't want your front-end web developers to have to worry about something like a pkid or how to retrieve this for each of the product APIs. To perform an import of a user, you really want them to be able to supply only the user ID. In your code on the backend, you can look up the pkid based on that user and perform the import if that user is found.

Step 3 - Import an LDAP User by User Id

To import an LDAP user using your API, you must implement the following logic:

  1. Read in the user id for the user you want to import.
  2. Use myCUPI's send_request to request "import/users/ldap" for the desired user, using the query=(alias is <userid>) parameter to look them up in the Unity Connection LDAP database: myCUPI._cupi_request("import/users/ldap", parameters=params), where params is the query.
  3. If you receive a user in the response, use that user's pkid to construct a POST request to "import/users/ldap", which imports the user, just as you had seen with Postman. E.g.: myCUPI._cupi_request("import/users/ldap", parameters=params, payload=args, http_method='POST'), where the parameter specifies the voicemail templateAlias.
  4. Handle errors, such as no user found or an error in querying for the user.

You can now implement this logic in the code.

  1. In your VS Code tab, open up flaskr/api/v1/cuc.py
  2. In the post() function in the cuc_user_api class, add the following:

        @api.expect(cuc_importldap_user_post_args, validate=True)
        def post(self, userid):
            """
            Import Unity Connection user from LDAP using the user ID / alias.
            """
            # Read arguments: ListInDirectory, IsVmEnrolled, PIN, and/or ResetMailbox
            args = cuc_importldap_user_post_args.parse_args(request)
            if 'templateAlias' not in args:
                args['templateAlias'] = 'voicemailusertemplate'
    
            # Look up pkid from user ID
            params = {'query': '(alias is {})'.format(userid)}
            user = myCUPI._cupi_request("import/users/ldap", parameters=params)
    
            # Either a single user was returned, no users were found, or an error occurred.
            try:
                if user['response']['@total'] == '1':
                    # Single user found.  Import the user using the pkid and settings
                    args['pkid'] = user['response']['ImportUser'][0]['pkid']
                    # The templateAlias needs to be a parameter, while the pkid and other settings are part of payload
                    params = {'templateAlias': args['templateAlias']}
                    return myCUPI._cupi_request("import/users/ldap", parameters=params, payload=args, http_method='POST')
    
                else:
                    # No users were found
                    return {'success': False,
                            'message': 'Found {} users to import with user id {}'.format(user['response']['@total'], userid),
                            'response': user['response']}
            except KeyError:
                pass
            # Return the errored user look up data
            return user

  3. Save this file.

Step 4 - Get a User by ID

The Unity Connection-related tasks such as modifying and deleting a user should require only the user ID as an input. For that, you need a function that can look up a Unity Connection user account and retrieve their pkid. Luckily, we have already seen that Unity Connection allows for a specific user search using a GET to /vmrest/users with a query parameter such as the following: ?query=(alias is <userid>). You should put this in its own function, because every one of the subsequent requests will need that exact same lookup.

  1. In your VS Code tab, open up flaskr/api/v1/cuc.py
  2. In the get_user_by_id() function, replace the pass line with the following:

    def get_user_by_id(userid):
        '''
        Get a voicemail user from the Unity Connection system by user alias.
    
        Reference:
        https://www.cisco.com/c/en/us/td/docs/voice_ip_comm/connection/REST-API/CUPI_API/b_CUPI-API/b_CUPI-API_chapter_0111.html#id_38638
        '''
    
        # Build the parameter dictionary for the request
        params = {'query': '(alias is {})'.format(userid)}
        user = myCUPI._cupi_request("users", parameters=params)
        # Verify there were no errors returned
        if user['success']:
            try:
                # If @total exists, it will be 1, otherwise a KeyError will be raised
                if user['response']['@total'] == '1':
                    # Retrieve the first (and only member of the users list)
                    return myCUPI._cupi_request('users/' + user['response']['User'][0]['ObjectId'])
            except KeyError:
                pass
        return user

  3. To test this later, add the following to the get() function:

    @api.route("/users/<userid>")
    @api.param('userid', 'The userid (alias) of the user')
    class cuc_user_api(Resource):
        def get(self, userid):
            """
            Get user from Unity Connection using the user ID / alias.
            """
            return get_user_by_id(userid)

This will search for and return a specific Unity Connection user, with all their settings, given only the user ID.

Step 5 - Modify User Settings and Voicemail PIN by User Id

Now that you have a way to look up a user account, you can then add the ability to modify them. You may also want to provide a simple flag for a user to reset a mailbox and change a user's PIN. We have handled gathering the input for you. Now, a few API calls in succession will be required to implement user modifications, based on what was supplied. Keep in mind that the credential/PIN changes are a different API method than other user settings.

The following logic will need to be implemented:

  1. Read in the arguments. This is done for you and will include the following dictionary keys:
    • ListInDirectory - True/False - controls whether the account is listed in the Unity directory
    • IsVmEnrolled - True/False - controls whether the account's initial enrollment prompt will play
    • PIN - Integer - a new PIN for the account
    • ResetMailbox - True/False - ability to reset the ResetMailbox and TimeHacked, which, in effect, allows the user to try to log in again after being locked out (which usually happens when a password was forgotten)
  2. Look up the user in order to get the account's ObjectId
  3. If there are user settings to change, issue the PUT request to /users/ObjectId to do so.
  4. If credential/voicemail reset settings were supplied then issue a PUT request for that (since it's a different URL, /users/ObjectId/credential/pin)
  5. You need to handle errors such as the user not found, or if one of these requests generates an error

Now turn your attention to implementing this in your code.

  1. In your VS Code tab, open up flaskr/api/v1/cuc.py
  2. In the put() function of the cuc_user_api class, replace the pass line with the following to check if any arguments were actually supplied and look up the user and then the user's object ID:

        @api.expect(cuc_users_put_args, validate=True)
        def put(self, userid):
            """
            Update user from Unity Connection using the user ID / alias.
            """
            # Need to distinguish which arguments go with which update, since they're different methods
            user_settings = ['ListInDirectory', 'IsVmEnrolled']
            pin_settings = ['PIN', 'ResetMailbox']
            # Read arguments: ListInDirectory, IsVmEnrolled, PIN, and/or ResetMailbox
            args = cuc_users_put_args.parse_args(request)
    
            # If no arguments were detected, there's nothing to do
            if len(args) > 0:
                # Look up the CUC user
                user = get_user_by_id(userid)
    
                if user['success']:
                    # Either a single user was returned, no users were found, or an error occurred.
                    try:
                        # Look up the user's object ID. If it fails, no user was found
                        user_id = user['response']['ObjectId']
    
                        # Check if any user settings were supplied
                        if any(user_setting in args for user_setting in user_settings):
                            payload = {}
                            for user_setting in user_settings:
                                if user_setting in args:
                                    payload[user_setting] = args[user_setting]
                            # Update the user's settings
                            user_result = myCUPI._cupi_request("users/" + user_id,
                                                               payload=payload, http_method='PUT')
    
                            # If the update failed, return; don't continue to try to update again with creds
                            if not user_result['success']:
                                return user_result
    
                        # Check if PIN or ResetMailbox were supplied
                        if any(pin_setting in args for pin_setting in pin_settings):
                            cred_payload = {}
                            if 'PIN' in args:
                                cred_payload['Credentials'] = args['PIN']
                            if 'ResetMailbox' in args:
                                # ResetMailbox is actually a matter of resetting HackCount and TImeHacked
                                cred_payload['HackCount'] = 0
                                cred_payload['TimeHacked'] = []
                            # Update the user's credentials
                            return myCUPI._cupi_request('users/' + str(user['response']['ObjectId']) + '/credential/pin',
                                                        payload=cred_payload, http_method='PUT')
                        return user_result
                    except KeyError:
                        # Zero users were found
                        return {'success': False,
                                'message': 'Found 0 users with user id {}'.format(userid),
                                'response': user['response']}
                else:
                    # Error in querying for the user
                    return user
            else:
                # No arguments supplied besides the userid
                return {'success': True, 'message': 'No changes specified for {}'.format(userid), 'response': ''}

  3. Save this file.

Step 6 - Delete the Voicemail Account by User Id

As with other userid operations, you must first look up the user in order to get its Object ID. Then, if there were no errors, issue the DELETE to remove the account.

  1. In your VS Code tab, open up flaskr/api/v1/cuc.py
  2. In the delete_user() function, replace the pass line with the following return line:

        def delete(self, userid):
            """
            Delete user from Unity Connection using the user ID / alias.
            """
            # Look up the CUC user
            user = get_user_by_id(userid)
    
            # Either a single user was returned, no users were found, or an error occurred.
            if user['success']:
                try:
                    # Single user found.  Delete the user using the object ID
                    user_id = user['response']['ObjectId']
                    return myCUPI._cupi_request("users/" + user_id, http_method='DELETE')
    
                except KeyError:
                    # No users were found
                    return {'success': False,
                            'message': 'Found 0 users to delete with user id {}'.format(userid),
                            'response': user['response']}
            else:
                # Error in querying for user
                return user

  3. Save this file.

Step 7 - Verification

You have added a lot of code so far, now it is time to make sure it works as expected.

  1. In VS Code, make sure flaskr/api/v1/cuc.py is saved
  2. Restart your Flask app by clicking the green restart button from the controller: which you should always see along the top of the window, when Flask is running.
    Or, if Flask is not started at all, click the Debug button on the left side followed by the green start/play button next to Start LTRCOL-2574 Portal
  3. If the Terminal window at bottom right shows: * Running on http://10.0.111.40.40:5000/ (Press CTRL+C to quit), then your Flask app is running.
  4. Access the Swagger UI page at http://dev1.pod11.col.lab:5000/api/v1/
  5. Expand the cuc section.
  6. Perform the individual checks:
    1. Get LDAP Users Available for Import
      1. Click the GET /cuc/ldap_users operation.
      2. On the right, click Try it out.
      3. In the expanded section, click Execute.
      4. Examine the response. You should note a Server response code of 200 and a response body with the users that can be imported.

        Make sure pod11user1 user appears in the list, or you won't be able to import that user
    2. Import an LDAP User by User Id
      1. Click the POST /cuc/users/{userid} operation.
      2. On the right, click Try it out.
      3. For the userid setting, enter pod11user1
      4. Click Execute.
      5. Examine the response. You should note a Server response code of 200 and a response body that should indicate success such as in this example:

            {
                "success": true,
                "message": "/vmrest/users/09d5120d-e678-4577-a936-121cf6910eb7"
                "response": ""
            }

        The response message, is just as you saw with Postman. The difference is that you don't really need to pay attention to the ID assigned, since your code can retrieve this any time using the user ID.
    3. Modify user settings and voicemail PIN by User Id
      1. Click the PUT /cuc/users/{userid} operation.
      2. On the right, click Try it out.
      3. Set the following:
        • For ListInDirectory, select false
        • For IsVmEnrolled, select false
        • For ResetMailbox, select true
        • Set userid to pod11user1
      4. Click Execute.
      5. Examine the response. You should note a Server response code of 200 to indicate a success.
    4. Get a User by ID
      1. Click the GET /cuc/users/{userid} option.
      2. On the right, click Try it out.
      3. Set userid to pod11user1
      4. Click Execute.
      5. Examine the response and make sure you received a response of 200. Check to see if ListInDirectory and IsVmEnrolled are now false.
    5. Delete the voicemail account by User Id
      1. Click the red DELETE /cuc/users/{userid} method.
      2. On the right, click Try it out.
      3. Set userid to pod11user1
      4. Click Execute.
      5. Examine the response. You should note a Server response code of 200 and a response body that shows success as true. To test a failure, try executing this again and you should see something like:

            {
                "success": false,
                "message": "Found 0 users to import with user id pod11user1",
                "response": {
                    "@total": "0"
                }
            }

You have a functional CUPI API! Now you are ready to implement the same for Cisco Meeting Server.


from flask import request
from flask_restx import Namespace, Resource
from flaskr.cuc.v1.cupi import CUPI
from flaskr.api.v1.parsers import cuc_importldap_user_post_args
from flaskr.api.v1.parsers import cuc_users_get_args
from flaskr.api.v1.parsers import cuc_users_put_args
from os import getenv

api = Namespace('cuc', description='Cisco Unity Connection APIs')

myCUPI = CUPI(getenv('CUC_HOSTNAME'), getenv('CUC_USERNAME'), getenv('CUC_PASSWORD'), port=getenv('CUC_PORT'))


@api.route("/version")
class cuc_version_api(Resource):
    def get(self):
        """
        Retrieves Unity Connection version.
        """
        return myCUPI._cupi_request("version/product/")

def get_search_params(args):
    """
    Returns CUC search parameters from request arguments as a dictionary.
    Builds a valid query parameter from column/match_type/search string.
    """
    params = {'sortorder': args.get('sortorder'),
              'rowsPerPage': args.get('rowsPerPage'),
              'pageNumber': args.get('pageNumber')}

    # Make sure if search is supplied that the 'query' is built properly
    if args.get('search') is not None:
        params['query'] = '({} {} {})'.format(args.get('column'), args.get('match_type'), args.get('search'))
    return params


@api.route("/ldap_users")
class cuc_import_ldapuser_api(Resource):
    @api.expect(cuc_users_get_args, validate=True)
    def get(self):
        """
        Retrieves LDAP users synched to Unity Connection.
        """
        # Read arguments: column, match_type, search, sortorder, rowsPerPage, pageNumber
        args = cuc_users_get_args.parse_args(request)
        params = get_search_params(args)

        return myCUPI._cupi_request("import/users/ldap", parameters=params)


def get_user_by_id(userid):
    '''
    Get a voicemail user from the Unity Connection system by user alias.

    Reference:
    https://www.cisco.com/c/en/us/td/docs/voice_ip_comm/connection/REST-API/CUPI_API/b_CUPI-API/b_CUPI-API_chapter_0111.html#id_38638
    '''

    # Build the parameter dictionary for the request
    params = {'query': '(alias is {})'.format(userid)}
    user = myCUPI._cupi_request("users", parameters=params)
    # Verify there were no errors returned
    if user['success']:
        try:
            # If @total exists, it will be 1, otherwise a KeyError will be raised
            if user['response']['@total'] == '1':
                # Retrieve the first (and only member of the users list)
                return myCUPI._cupi_request('users/' + user['response']['User'][0]['ObjectId'])
        except KeyError:
            pass
    return user


@api.route("/users/<userid>")
@api.param('userid', 'The userid (alias) of the user')
class cuc_user_api(Resource):
    def get(self, userid):
        """
        Get user from Unity Connection using the user ID / alias.
        """
        return get_user_by_id(userid)

    @api.expect(cuc_importldap_user_post_args, validate=True)
    def post(self, userid):
        """
        Import Unity Connection user from LDAP using the user ID / alias.
        """
        # Read arguments: ListInDirectory, IsVmEnrolled, PIN, and/or ResetMailbox
        args = cuc_importldap_user_post_args.parse_args(request)
        if 'templateAlias' not in args:
            args['templateAlias'] = 'voicemailusertemplate'

        # Look up pkid from user ID
        params = {'query': '(alias is {})'.format(userid)}
        user = myCUPI._cupi_request("import/users/ldap", parameters=params)

        # Either a single user was returned, no users were found, or an error occurred.
        try:
            if user['response']['@total'] == '1':
                # Single user found.  Import the user using the pkid and settings
                args['pkid'] = user['response']['ImportUser'][0]['pkid']
                # The templateAlias needs to be a parameter, while the pkid and other settings are part of payload
                params = {'templateAlias': args['templateAlias']}
                return myCUPI._cupi_request("import/users/ldap", parameters=params, payload=args, http_method='POST')

            else:
                # No users were found
                return {'success': False,
                        'message': 'Found {} users to import with user id {}'.format(user['response']['@total'], userid),
                        'response': user['response']}
        except KeyError:
            pass
        # Return the errored user look up data
        return user

    @api.expect(cuc_users_put_args, validate=True)
    def put(self, userid):
        """
        Update user from Unity Connection using the user ID / alias.
        """
        # Need to distinguish which arguments go with which update, since they're different methods
        user_settings = ['ListInDirectory', 'IsVmEnrolled']
        pin_settings = ['PIN', 'ResetMailbox']
        # Read arguments: ListInDirectory, IsVmEnrolled, PIN, and/or ResetMailbox
        args = cuc_users_put_args.parse_args(request)

        # If no arguments were detected, there's nothing to do
        if len(args) > 0:
            # Look up the CUC user
            user = get_user_by_id(userid)

            if user['success']:
                # Either a single user was returned, no users were found, or an error occurred.
                try:
                    # Look up the user's object ID. If it fails, no user was found
                    user_id = user['response']['ObjectId']

                    # Check if any user settings were supplied
                    if any(user_setting in args for user_setting in user_settings):
                        payload = {}
                        for user_setting in user_settings:
                            if user_setting in args:
                                payload[user_setting] = args[user_setting]
                        # Update the user's settings
                        user_result = myCUPI._cupi_request("users/" + user_id,
                                                           payload=payload, http_method='PUT')

                        # If the update failed, return; don't continue to try to update again with creds
                        if not user_result['success']:
                            return user_result

                    # Check if PIN or ResetMailbox were supplied
                    if any(pin_setting in args for pin_setting in pin_settings):
                        cred_payload = {}
                        if 'PIN' in args:
                            cred_payload['Credentials'] = args['PIN']
                        if 'ResetMailbox' in args:
                            # ResetMailbox is actually a matter of resetting HackCount and TImeHacked
                            cred_payload['HackCount'] = 0
                            cred_payload['TimeHacked'] = []
                        # Update the user's credentials
                        return myCUPI._cupi_request('users/' + str(user['response']['ObjectId']) + '/credential/pin',
                                                    payload=cred_payload, http_method='PUT')
                    return user_result
                except KeyError:
                    # Zero users were found
                    return {'success': False,
                            'message': 'Found 0 users with user id {}'.format(userid),
                            'response': user['response']}
            else:
                # Error in querying for the user
                return user
        else:
            # No arguments supplied besides the userid
            return {'success': True, 'message': 'No changes specified for {}'.format(userid), 'response': ''}

    def delete(self, userid):
        """
        Delete user from Unity Connection using the user ID / alias.
        """
        # Look up the CUC user
        user = get_user_by_id(userid)

        # Either a single user was returned, no users were found, or an error occurred.
        if user['success']:
            try:
                # Single user found.  Delete the user using the object ID
                user_id = user['response']['ObjectId']
                return myCUPI._cupi_request("users/" + user_id, http_method='DELETE')

            except KeyError:
                # No users were found
                return {'success': False,
                        'message': 'Found 0 users to delete with user id {}'.format(userid),
                        'response': user['response']}
        else:
            # Error in querying for user
            return user