A detailed, albeit contrived, example of how to use PyXB is in Generating Binding Classes.
Multiple real-world XML and web service examples of varying complexity are provided in the examples subdirectories of the PyXB distribution and of the various bundles. Some are WSDL services, and others are simply XMLSchema documents. Often there are a couple trivial programs that show how the bindings are used. The script test.sh in each directory can be used to generate the bindings and run the programs in a single step.
See the README.txt file in each example subdirectory for more information.
You may also want to look at some of the unit tests for other ideas.
The Dictionary web service at Aonaware.
The example define.py looks up a word in all available dictionaries:
import dict
import urllib2
import pyxb.utils.domutils as domutils
import sys
word = 'xml'
if 1 < len(sys.argv):
word = sys.argv[1]
# Create a REST-style query to retrieve the information about this dictionary.
uri = 'http://services.aonaware.com/DictService/DictService.asmx/Define?word=%s' % (word,)
rxml = urllib2.urlopen(uri).read()
resp = dict.CreateFromDOM(domutils.StringToDOM(rxml))
print 'Definitions of %s:' % (resp.Word,)
for definition in resp.Definitions.Definition:
print 'From %s (%s):' % (definition.Dictionary.Name, definition.Dictionary.Id)
print definition.WordDefinition
print
The example showdict.py lists the available dictionaries:
import dict
import urllib2
import pyxb.utils.domutils as domutils
from xml.dom import minidom
# Get the list of dictionaries available from the service.
port_uri = 'http://services.aonaware.com/DictService/DictService.asmx'
uri = port_uri + '/DictionaryList'
dle_xml = urllib2.urlopen(uri).read()
dle_dom = domutils.StringToDOM(dle_xml)
dle = dict.ArrayOfDictionary.createFromDOM(dle_dom)
op_path = '/DictionaryInfo'
for d in dle.Dictionary:
# Create a REST-style query to retrieve the information about this dictionary.
uri = '%s%s?dictId=%s' % (port_uri, op_path, d.Id)
resp = urllib2.urlopen(uri).read()
# The response is a simple type derived from string, so we can
# just extract and print it.
di_resp = dict.CreateFromDOM(domutils.StringToDOM(resp))
# Do the "encode" garbage because one of these dictionaries has a
# non-ASCII character
print "%s (%s)\n%s\n" % (d.Name.encode('utf-8'), d.Id.encode('utf-8'), di_resp.encode('utf-8'))
Sorry, no automatic generation of classes corresponding to the WSDL operations. Next release, maybe.
A free weather service. The REST interface was demonstrated as the Thirty Second Example. A SOAP interface example is in client.py:
import weather
import time
import sys
import pyxb.bundles.wssplat.soap11 as soapenv
import urllib2
zip = 55113
if 1 < len(sys.argv):
zip = int(sys.argv[1])
# Create an envelope, and give it a body that is the request for the
# service we want.
env = soapenv.Envelope(soapenv.Body(weather.GetCityForecastByZIP(ZIP=str(zip))))
file('request.xml', 'w').write(env.toxml("utf-8"))
# Invoke the service
uri = urllib2.Request('http://wsf.cdyne.com/WeatherWS/Weather.asmx',
env.toxml("utf-8"),
{ 'SOAPAction' : "http://ws.cdyne.com/WeatherWS/GetCityForecastByZIP", 'Content-Type': 'text/xml' } )
rxml = urllib2.urlopen(uri).read()
file('response.xml', 'w').write(rxml)
# Convert the response to a SOAP envelope, then extract the actual
# response from the wildcard elements of the body. Note that because
# the weather namespace was registered, PyXB already created the
# binding for the response.
soap_resp = soapenv.CreateFromDocument(rxml)
resp = soap_resp.Body.wildcardElements()[0]
fc_return = resp.GetCityForecastByZIPResult
if fc_return.Success:
print 'Got response for %s, %s:' % (fc_return.City, fc_return.State)
for fc in fc_return.ForecastResult.Forecast:
when = time.strftime('%A, %B %d %Y', fc.Date.timetuple())
outlook = fc.Desciption # typos in WSDL left unchanged
low = fc.Temperatures.MorningLow
high = fc.Temperatures.DaytimeHigh
print ' %s: %s, from %s to %s' % (when, outlook, low, high)
Note the various misspellings in the schema (e.g., “Desciption”). Also, be aware that the weather information in this service does not get updated often, and sometimes fails to provide valid dates. Try various zip codes; usually you can find one that works.
Interact with the National Digital Forecast Database.
Use the genbindings.sh script to retrieve the schema for Digital Weather Markup Language and generate the bindings. Note that the schema has two levels of include directives, which PyXB follows.
The examples for this service are too long to include into the web documentation. forecast.py uses the REST interface to get the forecast temperature data for two locations, and print it in several ways. latlon.py does something similar but for a latitude/longitude pair, using SOAP, and requesting more data.
A commercial service for television listings. Only one sample document is available for testing; it is retrieved using genbindings.sh. The dumpsample.py demonstrates extending a binding to add a custom method, and parsing content with both DOM and SAX. It also provides timing information; the document is about 200KB, and takes several seconds to parse.
import tmstvd
import pyxb.utils.domutils as domutils
import xml.dom.minidom
import pyxb.utils.saxdom
import pyxb.binding.saxer
import time
#import cProfile
# Extend the anonymous class used by the xtvd element to add a method
# we can use to test equality of two instances. Normally, you'd just
# refer to the complex type binding class itself, but we don't know
# what PyXB named it.
class my_xtvd (tmstvd.xtvd.typeDefinition()):
def equal (self, other, verbose=False):
if len(self.stations.station) != len(other.stations.station):
return False
for i in range(len(self.stations.station)):
s = self.stations.station[i]
o = other.stations.station[i]
if (s.callSign != o.callSign) or (s.name != o.name) or (s.id != o.id):
return False
if verbose:
print 'Match station %s is %s, id %d' % (s.callSign, s.name, s.id)
return True
tmstvd.xtvd.typeDefinition()._SetSupersedingClass(my_xtvd)
# The sample document.
xml_file = 'tmsdatadirect_sample.xml'
print 'Generating binding from %s with minidom' % (xml_file,)
mt1 = time.time()
xmls = open(xml_file).read()
mt2 = time.time()
dom = xml.dom.minidom.parseString(xmls)
mt3 = time.time()
#cProfile.run('tmstvd.CreateFromDOM(dom.documentElement)', 'dom.prf')
dom_instance = tmstvd.CreateFromDOM(dom.documentElement)
print 'minidom first callSign at %s' %(dom_instance.stations.station[0].callSign._location(),)
mt4 = time.time()
print 'Generating binding from %s with SAXDOM' % (xml_file,)
dt1 = time.time()
dom = pyxb.utils.saxdom.parseString(xmls, location_base=xml_file)
dt2 = time.time()
#cProfile.run('tmstvd.CreateFromDOM(dom.documentElement)', 'saxdom.prf')
saxdom_instance = tmstvd.CreateFromDOM(dom.documentElement)
print 'SAXDOM first callSign at %s' % (saxdom_instance.stations.station[0].callSign._location(),)
dt3 = time.time()
print 'Generating binding from %s with SAX' % (xml_file,)
st1 = time.time()
saxer = pyxb.binding.saxer.make_parser(location_base=xml_file)
handler = saxer.getContentHandler()
st2 = time.time()
saxer.parse(open(xml_file))
#cProfile.run('saxer.parse(open(xml_file))', 'sax.prf')
st3 = time.time()
sax_instance = handler.rootObject()
print 'SAXER first callSign at %s' % (sax_instance.stations.station[0].callSign._location(),)
print 'DOM-based read %f, parse %f, bind %f, total %f' % (mt2-mt1, mt3-mt2, mt4-mt3, mt4-mt2)
print 'SAXDOM-based parse %f, bind %f, total %f' % (dt2-dt1, dt3-dt2, dt3-dt1)
print 'SAX-based read %f, parse and bind %f, total %f' % (st2-st1, st3-st2, st3-st1)
print "Equality test on DOM vs SAX: %s" % (dom_instance.equal(sax_instance),)
print "Equality test on SAXDOM vs SAX: %s" % (saxdom_instance.equal(sax_instance, verbose=True),)
This service provides the latitude and longitude for free-form US addresses. It also demonstrates several of the pitfalls of using WSDL, which has a very lax concept of schemas, with a system that expects to operate on validatable documents. The following changes were made to make the service easier to work with:
- Change the element form default to qualified. This is necessary because there is a non-absent target namespace in the schema, and the documents returned from the service set it as the default namespace. This causes the XML engine to associate that namespace with locally-scoped elements like “number”, while the original unqualified form default caused the schema to record them with no namespace.
- Set minOccurs on all the elements, since some are missing from some responses
- Set nillable on all elements that are observed to use xsi:nil="true" in response documents
- Provide types and elements corresponding to the request and response messages, since PyXB’s WSDL support does not currently generate them from the operation messages.
genbindings.sh applies changes to the WSDL after retrieving it and prior to generating the bindings.
A second complication is the need to burrow down through wildcard elements in the binding instance generated from the SOAP response. This is the consequence of an error in the WSDL specification, which was discovered after too many tools had already worked around it. Currently, PyXB also requires that you work around it manually, although a customization of the relevant SOAP encoding class could make it unnecessary.
import GeoCoder
from pyxb import BIND
import sys
import pyxb.utils.domutils as domutils
import pyxb.bundles.wssplat.soap11 as soapenv
import pyxb.bundles.wssplat.soapenc as soapenc
import urllib2
address = '1600 Pennsylvania Ave., Washington, DC'
if 1 < len(sys.argv):
address = sys.argv[1]
env = soapenv.Envelope(Body=BIND(GeoCoder.geocode(address)))
uri = urllib2.Request('http://rpc.geocoder.us/service/soap/',
env.toxml("utf-8"),
{ 'SOAPAction' : "http://rpc.geocoder.us/Geo/Coder/US#geocode", 'Content-Type': 'text/xml' } )
rxml = urllib2.urlopen(uri).read()
#file('response.xml', 'w').write(rxml)
#rxml = file('response.xml').read()
response = soapenv.CreateFromDocument(rxml)
# OK, here we get into ugliness due to WSDL's concept of schema in the
# SOAP encoding not being consistent with XML Schema, even though it
# uses the same namespace. See
# http://tech.groups.yahoo.com/group/soapbuilders/message/5879. In
# short, the WSDL spec shows an example using soapenc:Array where a
# restriction was used to set the value of the wsdl:arrayType
# attribute. This restriction failed to duplicate the element content
# of the base type, resulting in a content type of empty in the
# restricted type. Consequently, PyXB can't get the information out
# of the DOM node, and we have to skip over the wildcard items to find
# something we can deal with.
# As further evidence the folks who designed SOAP 1.1 didn't know what
# they were doing, the encodingStyle attribute that's supposed to go
# in the Envelope can't validly be present there, since it's not
# listed and it's not in the namespace admitted by the attribute
# wildcard. Fortunately, PyXB doesn't currently validate wildcards.
encoding_style = response.wildcardAttributeMap().get(soapenv.Namespace.createExpandedName('encodingStyle'))
items = []
if encoding_style == soapenc.Namespace.uri():
gcr = response.Body.wildcardElements()[0]
soap_array = gcr.wildcardElements()[0]
items = soap_array.wildcardElements()
else:
pass
for item in items:
if (item.lat is None) or item.lat._isNil():
print 'Warning: Address did not resolve'
print '''
%s %s %s %s %s
%s, %s %s
%s %s''' % (item.number, item.prefix, item.street, item.type, item.suffix,
item.city, item.state, item.zip,
item.lat, item.long)
See the directory examples/OpenGIS in the distribution.