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!