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:
_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:
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)
If the results content is just a string, then return it. This is the situation you observed
in the response of a POST message.
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:
Get Version
Get LDAP Users Available for Import
Import an LDAP User by User Id
Get a User by ID
Modify user settings and voicemail PIN by User Id
Delete the voicemail account by User Id
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.
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'))
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.
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.
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.
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:
Read in the user id for the user you want to import.
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.
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.
Handle errors, such as no user found or an error in querying for the user.
In the post() function in the cuc_user_api
class, add the following:
@api.expect(cuc_importldap_user_post_args, validate=True)defpost(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'notin 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 foundreturn{'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 datareturn user
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.
In the get_user_by_id() function, replace the pass
line with the following:
defget_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 returnedif user['success']:try:# If @total exists, it will be 1, otherwise a KeyError will be raisedif 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:passreturn user
To test this later, add the following to the get() function:
@api.route("/users/<userid>")@api.param('userid','The userid (alias) of the user')classcuc_user_api(Resource):defget(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:
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)
Look up the user in order to get the account's ObjectId
If there are user settings to change, issue the PUT request to /users/ObjectId to do so.
If credential/voicemail reset settings were supplied then issue a PUT
request for that (since it's a different URL, /users/ObjectId/credential/pin)
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.
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)defput(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 doiflen(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 suppliedifany(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 credsifnot user_result['success']:return user_result
# Check if PIN or ResetMailbox were suppliedifany(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 credentialsreturn myCUPI._cupi_request('users/'+str(user['response']['ObjectId'])+'/credential/pin',
payload=cred_payload, http_method='PUT')return user_result
except KeyError:# Zero users were foundreturn{'success':False,'message':'Found 0 users with user id {}'.format(userid),'response': user['response']}else:# Error in querying for the userreturn user
else:# No arguments supplied besides the useridreturn{'success':True,'message':'No changes specified for {}'.format(userid),'response':''}
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.
In the delete_user() function, replace the pass
line with the following return line:
defdelete(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 foundreturn{'success':False,'message':'Found 0 users to delete with user id {}'.format(userid),'response': user['response']}else:# Error in querying for userreturn user
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.
In VS Code, make sure flaskr/api/v1/cuc.py is saved
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
If the Terminal window at bottom right shows: * Running on http://10.0.101.40.40:5000/ (Press CTRL+C to quit), then
your Flask app is running.
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.
Modify user settings and voicemail PIN by User Id
Click the PUT /cuc/users/{userid} operation.
On the right, click Try it out.
Set the following:
For ListInDirectory, select false
For IsVmEnrolled, select false
For ResetMailbox, select true
Set userid to
pod1user1
Click Execute.
Examine the response. You should note a Server response code of
200 to indicate a success.
Get a User by ID
Click the GET /cuc/users/{userid} option.
On the right, click Try it out.
Set userid to
pod1user1
Click Execute.
Examine the response and make sure you received a response of
200. Check to see if ListInDirectory
and IsVmEnrolled are now false.
Delete the voicemail account by User Id
Click the red DELETE /cuc/users/{userid} method.
On the right, click Try it out.
Set userid to
pod1user1
Click Execute.
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 pod1user1","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")classcuc_version_api(Resource):defget(self):"""
Retrieves Unity Connection version.
"""return myCUPI._cupi_request("version/product/")defget_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 properlyif args.get('search')isnotNone:
params['query']='({} {} {})'.format(args.get('column'), args.get('match_type'), args.get('search'))return params
@api.route("/ldap_users")classcuc_import_ldapuser_api(Resource):@api.expect(cuc_users_get_args, validate=True)defget(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)defget_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 returnedif user['success']:try:# If @total exists, it will be 1, otherwise a KeyError will be raisedif 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:passreturn user
@api.route("/users/<userid>")@api.param('userid','The userid (alias) of the user')classcuc_user_api(Resource):defget(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)defpost(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'notin 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 foundreturn{'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 datareturn user
@api.expect(cuc_users_put_args, validate=True)defput(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 doiflen(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 suppliedifany(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 credsifnot user_result['success']:return user_result
# Check if PIN or ResetMailbox were suppliedifany(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 credentialsreturn myCUPI._cupi_request('users/'+str(user['response']['ObjectId'])+'/credential/pin',
payload=cred_payload, http_method='PUT')return user_result
except KeyError:# Zero users were foundreturn{'success':False,'message':'Found 0 users with user id {}'.format(userid),'response': user['response']}else:# Error in querying for the userreturn user
else:# No arguments supplied besides the useridreturn{'success':True,'message':'No changes specified for {}'.format(userid),'response':''}defdelete(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 foundreturn{'success':False,'message':'Found 0 users to delete with user id {}'.format(userid),'response': user['response']}else:# Error in querying for userreturn user