I have a buddy who asked for some help in constructing an SQL query to get all phones from CUCM that used a VPN Profile, which were configured in his system in one of several Common Phone Configuration elements. I helped him construct this:
SELECT d.name FROM device d
INNER JOIN commonphoneconfig cpc ON d.fkcommonphoneconfig=cpc.pkid
INNER JOIN vpnprofile vp ON cpc.fkvpnprofile = vp.pkid
WHERE vp.name = ‘vpn-profile-name’
However, he wanted then to be able to run this using python, as he wanted to match this up against registered phones on the system to analyse active IP registrations to determine which phones were remotely registered.
Although he’d done a rudimentary setup with suds, he wanted to migrate this to use python-zeep.
from collections import OrderedDict from zeep import Client from zeep.cache import SqliteCache from zeep.transports import Transport from zeep.plugins import HistoryPlugin from zeep.exceptions import Fault from zeep.helpers import serialize_object from requests import Session from requests.auth import HTTPBasicAuth from urllib3 import disable_warnings from urllib3.exceptions import InsecureRequestWarning from lxml import etree disable_warnings(InsecureRequestWarning) USERNAME = 'administrator' PASSWORD = 'ciscopsdt' FQDN = '10.10.20.1' WSDL = 'file://C://path//to//AXLAPI.wsdl' BINDING_NAME = "{http://www.cisco.com/AXLAPIService/}AXLAPIBinding" ADDRESS = "https://{fqdn}:8443/axl/".format(fqdn=FQDN) def element_list_to_ordered_dict(elements): return [OrderedDict((element.tag, element.text) for element in row) for row in elements] def sql_query(service, sql_statement): axl_resp = service.executeSQLQuery(sql=sql_statement) try: return element_list_to_ordered_dict(serialize_object(axl_resp)["return"]["rows"]) except KeyError: # single tuple response return element_list_to_ordered_dict( serialize_object(axl_resp)["return"]["row"]) except TypeError: # no SQL tuples return serialize_object(axl_resp)["return"] def sql_update(service, sql_statement): axl_resp = service.executeSQLUpdate(sql=sql_statement) return serialize_object(axl_resp)["return"]["rowsUpdated"] def main(): session = Session() session.verify = False session.auth = HTTPBasicAuth(USERNAME, PASSWORD) transport = Transport(cache=SqliteCache(), session=session, timeout=60) history = HistoryPlugin() client = Client(wsdl=WSDL, transport=transport, plugins=[history]) axl = client.create_service(BINDING_NAME, ADDRESS) sql_statement = "SELECT d.name " \ "FROM device d " \ "INNER JOIN commonphoneconfig cpc ON d.fkcommonphoneconfig=cpc.pkid " \ "INNER JOIN vpnprofile vp ON cpc.fkvpnprofile = vp.pkid " \ "WHERE vp.name = 'vpn-profile-name'" try: print(sql_query(service=axl, sql_statement=sql_statement)) except Fault as thin_axl_error: print(thin_axl_error.message) finally: try: for hist in [history.last_sent, history.last_received]: print(etree.tostring(hist["envelope"], encoding="unicode", pretty_print=True)) except (IndexError, TypeError): # catch cases where it fails before being put on the wire pass if __name__ == '__main__': main()
I’ve included some helper functions for parsing the xml and for handling the AXL output for both query
and update
cases. I’ve also included a helpful xml dump for troubleshooting.
Happy coding!
hi, thanks for this. just wonder on line 35 return element_list_to_ordered_dict(serialize_object(axl_resp)[“return”][“rows”])
shouldnt be row without s ?
LikeLiked by 1 person
Nope, there are different cases. That’s why its wrapped in a try/except construct…
LikeLike
Hi, Thanks for your post. it is very useful for me!
LikeLike
It’s a pleasure – glad you found it useful.
LikeLike
Hello Jonathan,
I am studying the program. I don’t understand about “element_list_to_ordered_dict(serialize_object(axl_resp)[“return”][“rows”])”. [“return”] and [“rows”] what do these two parameters do?
LikeLike
Hi Yue,
The response changes depending on what you get back from AXL. If there’s one SQL row returned, the XML response is of the form . If multiples, its . There’s also a case were there’s no row returned at all.
The try/except code parses this low-level stuff.
LikeLike
You can also get more information on serialize_object here:
https://docs.python-zeep.org/en/master/helpers.html
LikeLike
Awesome, I remember having to deal with this with Ruby, checking the type of the serialized return to be either hash, or array of hashes, etc. Moving to Python I see this is still the case but it’s pretty simple to handle. Great post.
LikeLiked by 1 person
Thanks. Glad it helped.
LikeLike
Hello
Can I use this code on my training video
LikeLiked by 1 person
also mention afterthenumber.com for the source
LikeLiked by 1 person
yes, you can use this, but please do mention the source.
LikeLike