1 # minisoap - minimal SOAP client

     2 #

     3 # Copyright (c) 2011, 2013  Radek Brich <radek.brich@devl.cz>

     4 #

     5 # Permission is hereby granted, free of charge, to any person obtaining a copy

     6 # of this software and associated documentation files (the "Software"), to deal

     7 # in the Software without restriction, including without limitation the rights

     8 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell

     9 # copies of the Software, and to permit persons to whom the Software is

    10 # furnished to do so, subject to the following conditions:

    11 #

    12 # The above copyright notice and this permission notice shall be included in

    13 # all copies or substantial portions of the Software.

    14 #

    15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR

    16 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,

    17 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE

    18 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER

    19 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,

    20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN

    21 # THE SOFTWARE.

    22 

    23 

    24 from http.client import HTTPConnection

    25 from urllib.parse import urlparse

    26 from lxml import etree

    27 from collections import OrderedDict

    28 

    29 debug = 0

    30 HTTPConnection.debuglevel = debug

    31 

    32 

    33 class SoapError(Exception):

    34     pass

    35 

    36 

    37 class SoapResponse(OrderedDict):

    38     def __init__(self, name):

    39         OrderedDict.__init__(self)

    40         self.name = name

    41 

    42     def _printout(self, out, res, level):

    43         indent = level * 2 * ' '

    44         for k in res.keys():

    45             if type(res[k]) is list:

    46                 for row in res[k]:

    47                     out.extend([indent, k, '\n'])

    48                     self._printout(out, row, level+1)

    49             else:

    50                 out.extend([indent, k, ': ', str(res[k]), '\n'])

    51 

    52     def __str__(self):

    53         out = [self.name, '\n']

    54         self._printout(out, self, 1)

    55         return ''.join(out).rstrip()

    56 

    57 

    58 class SoapClient:

    59     def __init__(self, url, namespace):

    60         self.url = url

    61         self.parsedurl = urlparse(url)

    62         self.namespace = namespace

    63         self.conn = HTTPConnection(self.parsedurl.netloc)

    64 

    65     def _build_soap_message(self, name, args):

    66         SOAP = 'http://schemas.xmlsoap.org/soap/envelope/'

    67         root = etree.Element('{%s}Envelope' % SOAP, nsmap={'soap':SOAP})

    68         body = etree.SubElement(root, '{%s}Body' % SOAP)

    69         func = etree.SubElement(body, '{%s}%s' % (self.namespace, name), nsmap={None:self.namespace})

    70         for key in args.keys():

    71             param = etree.SubElement(func, key)

    72             param.text = str(args[key])

    73         if debug:

    74             print(etree.tostring(root, pretty_print=True).decode('utf8'))

    75         return root

    76 

    77     def _break_params(self, func, d):

    78         for param in func.iterchildren():

    79             if len(param) == 0:

    80                 d[param.tag] = param.text

    81             else:

    82                 if not param.tag in d:

    83                     d[param.tag] = []

    84                 newd = OrderedDict()

    85                 self._break_params(param, newd)

    86                 d[param.tag].append(newd)

    87 

    88     def _break_soap_message(self, root):

    89         if not root.tag.endswith('}Envelope'):

    90             raise SoapError('Reply has no Envelope.')

    91 

    92         SOAP = root.tag.split('}')[0][1:]

    93 

    94         body_list = list(root.iterchildren('{%s}Body' % SOAP))

    95         if len(body_list) != 1:

    96             raise SoapError('Bad SOAP format (%d Body elements)' % len(body_list))

    97         body = body_list[0]

    98 

    99         func_list =list(body.iterchildren())

   100         if len(func_list) != 1:

   101             raise SoapError('Bad SOAP format (Body has %d children, should have one)' % len(body_list))

   102         func = func_list[0]

   103 

   104         if debug:

   105             print(etree.tostring(func, pretty_print=True).decode('utf8'))

   106         res = SoapResponse(func.tag)

   107         self._break_params(func, res)

   108 

   109         return res

   110 

   111     def _serialize_xml(self, root):

   112         return etree.tostring(root,

   113             pretty_print=True,

   114             encoding='utf-8',

   115             xml_declaration=True)

   116 

   117     def _parse_xml(self, file):

   118         tree = etree.parse(file)

   119         return tree.getroot()

   120 

   121     def _send_soap(self, body):

   122         headers = {

   123             'Content-type': "text/xml;charset='utf-8'",

   124             'Soapaction': "''",

   125             'Accept': 'text/xml'}

   126         self.conn.request('POST', self.parsedurl.path, body, headers)

   127         resp = self.conn.getresponse()

   128         if resp.status != 200:

   129             raise SoapError('Server replied %d %s' % (resp.status, resp.reason))

   130         return resp

   131 

   132     def __getattr__(self, name):

   133         def fnc(**kwargs):

   134             root = self._build_soap_message(name, kwargs)

   135             body = self._serialize_xml(root)

   136             resp = self._send_soap(body)

   137             root = self._parse_xml(resp)

   138             return self._break_soap_message(root)

   139 

   140         return fnc

   141 

   142 

   143 if __name__ == '__main__':

   144     service = SoapClient('http://10.0.0.1:34567/', namespace='http://devl.cz/')

   145     res = service.my_test_request()

   146     print(res)

   147