Viewing file:  soap_proxy.py (4.46 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# Licensed under the MIT license
# http://opensource.org/licenses/mit-license.php
# Copyright 2007 - Frank Scholz <coherence@beebits.net>
from twisted.python import failure
from coherence import log
from coherence.extern.et import ET, namespace_map_update
from coherence.upnp.core.utils import getPage, parse_xml
from coherence.upnp.core import soap_lite
class SOAPProxy(log.Loggable):
""" A Proxy for making remote SOAP calls.
Based upon twisted.web.soap.Proxy and
extracted to remove the SOAPpy dependency
Pass the URL of the remote SOAP server to the constructor.
Use proxy.callRemote('foobar', 1, 2) to call remote method
'foobar' with args 1 and 2, proxy.callRemote('foobar', x=1)
will call foobar with named argument 'x'.
"""
logCategory = 'soap'
def __init__(self, url, namespace=None, envelope_attrib=None, header=None, soapaction=None):
self.url = url
self.namespace = namespace
self.header = header
self.action = None
self.soapaction = soapaction
self.envelope_attrib = envelope_attrib
def callRemote(self, soapmethod, arguments):
soapaction = soapmethod or self.soapaction
if '#' not in soapaction:
soapaction = '#'.join((self.namespace[1],soapaction))
self.action = soapaction.split('#')[1]
self.info("callRemote %r %r %r %r", self.soapaction, soapmethod, self.namespace, self.action)
headers = { 'content-type': 'text/xml ;charset="utf-8"',
'SOAPACTION': '"%s"' % soapaction,}
if arguments.has_key('headers'):
headers.update(arguments['headers'])
del arguments['headers']
payload = soap_lite.build_soap_call("{%s}%s" % (self.namespace[1], self.action), arguments,
encoding=None)
self.info("callRemote soapaction: ", self.action,self.url)
self.debug("callRemote payload: ", payload)
def gotError(error, url):
self.warning("error requesting url %r" % url)
self.debug(error)
try:
tree = parse_xml(error.value.response)
body = tree.find('{http://schemas.xmlsoap.org/soap/envelope/}Body')
return failure.Failure(Exception("%s - %s" % (body.find('.//{urn:schemas-upnp-org:control-1-0}errorCode').text,
body.find('.//{urn:schemas-upnp-org:control-1-0}errorDescription').text)))
except:
import traceback
self.debug(traceback.format_exc())
return error
return getPage(self.url, postdata=payload, method="POST",
headers=headers
).addCallbacks(self._cbGotResult, gotError, None, None, [self.url], None)
def _cbGotResult(self, result):
#print "_cbGotResult 1", result
page, headers = result
#result = SOAPpy.parseSOAPRPC(page)
#print "_cbGotResult 2", result
def print_c(e):
for c in e.getchildren():
print c, c.tag
print_c(c)
self.debug("result: %r" % page)
tree = parse_xml(page)
#print tree, "find %s" % self.action
#root = tree.getroot()
#print_c(root)
body = tree.find('{http://schemas.xmlsoap.org/soap/envelope/}Body')
#print "body", body
response = body.find('{%s}%sResponse' % (self.namespace[1], self.action))
if response == None:
""" fallback for improper SOAP action responses """
response = body.find('%sResponse' % self.action)
self.debug("callRemote response ", response)
result = {}
if response != None:
for elem in response:
result[elem.tag] = self.decode_result(elem)
#print "_cbGotResult 3", result
return result
def decode_result(self, element):
type = element.get('{http://www.w3.org/1999/XMLSchema-instance}type')
if type is not None:
try:
prefix, local = type.split(":")
if prefix == 'xsd':
type = local
except ValueError:
pass
if type == "integer" or type == "int":
return int(element.text)
if type == "float" or type == "double":
return float(element.text)
if type == "boolean":
return element.text == "true"
return element.text or ""
|