Serializing Thin AXL SQL Query Responses with python-zeep

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!

Advertisements

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Google+ photo

You are commenting using your Google+ account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

w

Connecting to %s