Article 7 - Setting Device Ownership by Primary Extension

This article covers setting device device ownership information by using the association between the telephone number allocated to the end user in the enduser table and the Directory Number assigned to the primary line of telephone devices in the device table.

SQL Query Used to Select User Data from enduser Table

The SQL query we will use to get user information from the enduser table is shown below.

SELECT pkid, userid, telephonenumber FROM enduser ORDER BY userid

The query selects the following information:

SQL Query Used to Set Device Ownership by Primary Extension

The SQL query we will use to set the device owner by primary extension is shown below.

UPDATE device SET fkenduser = "fkenduser" WHERE pkid IN (SELECT fkdevice FROM devicenumplanmap WHERE fknumplan = (SELECT pkid FROM numplan WHERE dnorpattern = "telephonenumber") AND numplanindex = 1) AND tkclass = '1'

The Python Script

The Python script shown below will retrieve the information listed below from the enduser table:

The script then loops through the user data and sets the Owner ID (fkenduser) of any device whose line 1 extension matches the user telephonenumber value.xxxxxxxxxxxxxxxx

The script comprises the following high level steps:

  1. Load information needed to connect to the target CUCM server from an ini file.
  2. Connect to the CUCM AXL API using the settings from the ini file and download pkid, userid and telephonenumber for all users in the enduser table - these are stored as an ordered list.
  3. Loop the the list of users obtained in the previous step and set the OwnerId of any device which has a Directory Number assigned to Line 1 that matches the user’s telephonenumber value to the fkenduser value of the user. As this loop executes the result of the update is written to a timestamped CSV file named using the format setownerbyprimaryextensiondatalog-yyyy-mm-dd-hh-mm-ss.csv
import sys
import requests
from urllib3 import disable_warnings
from urllib3.exceptions import InsecureRequestWarning
from configparser import ConfigParser
import datetime
import pprint
import xmltodict
import csv
from csv import DictReader

disable_warnings(InsecureRequestWarning)


########################################################################################################################

def axlgetcookies(serveraddress, version, axluser, axlpassword):
    try:
        # Make AXL query using Requests module
        axlgetcookiessoapresponse = requests.get(f'https://{ipaddr}:8443/axl/', headers=soapheaders, verify=False,\
                                                 auth=(axluser, axlpassword), timeout=3)
        print(axlgetcookiessoapresponse)
        getaxlcookiesresult = axlgetcookiessoapresponse.cookies

        if '200' in str(axlgetcookiessoapresponse):
            # request is successful
            print('AXL Request Successful')
        elif '401' in str(axlgetcookiessoapresponse):
            # request fails due to invalid credentials
            print('Response is 401 Unauthorised - please check credentials')
        else:
            # request fails due to other cause
            print('Request failed! - HTTP Response Code is ' + axlgetcookiessoapresponse)

    except requests.exceptions.Timeout:
        axlgetcookiesresult = 'Error: IP address not found!'

    except requests.exceptions.ConnectionError:
        axlgetcookiesresult = 'Error: DNS lookup failed!'

    return getaxlcookiesresult

########################################################################################################################

def axlgetenduserdata(cookies):
    # Set SOAP request body
    axlgetenduserdatasoaprequest = f'<?xml version="1.0" encoding="UTF-8"?>\
    <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"\
     xmlns:ns="http://www.cisco.com/AXL/API/{version}"><soapenv:Header/><soapenv:Body><ns:executeSQLQuery><sql>\
    SELECT pkid, userid, telephonenumber FROM enduser ORDER BY userid</sql></ns:executeSQLQuery></soapenv:Body>\
       </soapenv:Envelope>'

    try:

        # Make SOAP request
        axlgetenduserdatasoapresponse = requests.post(f'https://{ipaddr}:8443/axl/',\
                                                          data= axlgetenduserdatasoaprequest, headers=soapheaders,\
                                                          verify=False, cookies=cookies, timeout=3)
        # Parse SOAP XML response to a dictionary
        axlgetenduserdatasoapresponse_dict = xmltodict.parse(axlgetenduserdatasoapresponse.text)
        # Remove unwanted parts of the dictionary to leave the data we require
        axlgetenduserdatasoapresponse_subdict = \
            axlgetenduserdatasoapresponse_dict['soapenv:Envelope']['soapenv:Body']['ns:executeSQLQueryResponse']\
                ['return']['row']
        pprint.pprint(axlgetenduserdatasoapresponse_subdict)

        return axlgetenduserdatasoapresponse_subdict

    except requests.exceptions.Timeout:
        print('IP address not found! ')
    except requests.exceptions.ConnectionError:
        print('DNS lookup failed!  ')


########################################################################################################################

def axlsetdeviceowner(cookies, fkenduser, telephonenumber):
    # Set SOAP request body
    axlsetdeviceownersoaprequest = f'<?xml version="1.0" encoding="UTF-8"?><soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:ns="http://www.cisco.com/AXL/API/{version}"><soapenv:Header/><soapenv:Body><ns:executeSQLUpdate><sql>UPDATE device SET fkenduser = "{fkenduser}" WHERE pkid IN (SELECT fkdevice FROM devicenumplanmap WHERE fknumplan = (SELECT pkid FROM numplan WHERE dnorpattern = "{telephonenumber}" AND fkroutepartition = (SELECT pkid FROM routepartition WHERE name = "{user_dn_partition}")) AND numplanindex = 1) AND tkclass = "1"</sql></ns:executeSQLUpdate></soapenv:Body></soapenv:Envelope>'

    try:

        # Make SOAP request
        axlsetdeviceownersoapresponse = requests.post(f'https://{ipaddr}:8443/axl/', data=axlsetdeviceownersoaprequest, headers=soapheaders, verify=False, cookies=cookies, timeout=9)

        # Parse SOAP XML response to a dictionary
        axlsetdeviceownersoapresponse_dict = xmltodict.parse(axlsetdeviceownersoapresponse.text)
        # pprint.pprint(axlsetdeviceownersoapresponse_dict)
        devices_updated = axlsetdeviceownersoapresponse_dict['soapenv:Envelope']['soapenv:Body']['ns:executeSQLUpdateResponse']['return']['rowsUpdated']
        return devices_updated

    except requests.exceptions.Timeout:
        print('IP address not found! ')
    except requests.exceptions.ConnectionError:
        print('DNS lookup failed!  ')


########################################################################################################################

def main():
    # Import values for the CUCM environment from ini file using ConfigParser

    global ipaddr, version, axluser, axlpassword, user_dn_partition, soapheaders, cookies
    parser = ConfigParser()
    parser.read('cucm-settings.ini')

    ipaddr = parser.get('CUCM', 'ipaddr')
    version = parser.get('CUCM', 'version')
    axluser = parser.get('CUCM', 'axluser')
    axlpassword = parser.get('CUCM', 'axlpassword')
    user_dn_partition = parser.get('CUCM', 'user_dn_partition')
    soapheaders = {'Content-type': 'text/xml', 'SOAPAction': 'CUCM:DB ver=' + version}


    # Get cookies for future AXL requests
    axlgetcookiesresult = axlgetcookies(ipaddr, version, axluser, axlpassword)
    print(axlgetcookiesresult)

    if 'JSESSIONIDSSO' in str(axlgetcookiesresult):
        print('Cookies Retrieved')

    else:
        sys.exit('Program has ended due to error!')

    # Get end user data via AXL function and read into list of lists
    enduserdata_dict = axlgetenduserdata(axlgetcookiesresult)
    #pprint.pprint(enduserdata_dict)

    # Set device ownership via AXL function using telephonenumber to primary extension mapping

    # Create filename for output file in format setownerbyprimaryextensiondatalog-year-month-dom-hour-minute-second.csv
    now = datetime.datetime.now()
    outputfile = 'setownerbyprimaryextensiondatalog-' + now.strftime("%Y-%m-%d-%H-%M-%S") + '.csv'

    # Open output log csv file
    setownerbyprimaryextensiondataoutputfile = open(outputfile, 'w', newline='')
    setownerbyprimaryextensiondataoutputwriter = csv.writer(setownerbyprimaryextensiondataoutputfile)
    setownerbyprimaryextensiondataoutputwriter.writerow(['fkenduser', 'userid', 'telephonenumber', 'userid', 'DevicesUpdated'])

    for item in enduserdata_dict:
        fkenduser = (item['pkid'])
        userid = (item['userid'])
        telephonenumber = (item['telephonenumber'])
        # print(f'{fkenduser}, {userid}, {telephonenumber}')

        if str(telephonenumber) != 'None':
            devices_updated = axlsetdeviceowner(axlgetcookiesresult, fkenduser, telephonenumber)
            print(f'{fkenduser}, {userid}, {telephonenumber}, {devices_updated}')
            setownerbyprimaryextensiondataoutputwriter.writerow([fkenduser, userid, telephonenumber, devices_updated])
        else:
            print(f'{fkenduser}, {userid}, {telephonenumber}, no phone number in user record')
            setownerbyprimaryextensiondataoutputwriter.writerow([fkenduser, userid, telephonenumber, '0'])

    setownerbyprimaryextensiondataoutputfile.close()
 

########################################################################################################################

if __name__ == "__main__":
    main()