example - soap request python 3

Generación de stubs soaplib de Python de WSDL

Question

Me gustaría generar una clase de servicio web SOAP stub utilizando el módulo soaplib de Python, basado en un WSDL existente. La idea es generar un simulacro para un servicio web de terceros.

¿Existe algún generador de códigos, o debemos escribir el nuestro?

Martín

Answer 1

Acabo de crear un repositorio github donde estoy mejorando el script de optilude para que funcione con soaplib2.0 y más. El enlace es https://github.com/fvieira/wsdl2soaplib .

Answer 2

De acuerdo, he intentado hackear mi script wsdl2interface ( http://pypi.python.org/pypi/wsdl2interface ) para generar el código soaplib. Creo que tengo algo que funciona, aunque no es bonito ni especialmente probado.

Lo pegaré aquí para el registro. Podría persuadirme para que lo libere si alguien lo necesita, aunque no es exactamente mi mejor código. Tenga en cuenta que utiliza el analizador WSDL de Suds para generar el código soaplib, que es un poco extraño en sí mismo.

Correr así:

$ wsdl2soaplib <url or filename of WSDL> > wsdl.py

El código (necesitarás espuma en tu camino, idealmente en virtualenv):

from StringIO import StringIO

import os.path
import sys
import textwrap
import keyword
import re

import suds.client

VALID_IDENTIFIER_RE                   = re.compile(r"[_A-Za-z][_A-Za-z1-9]*")
VALID_IDENTIFIER_FIRST_LETTER_RE      = re.compile(r"[_A-Za-z]")
VALID_IDENTIFIER_SUBSEQUENT_LETTER_RE = re.compile(r"[_A-Za-z1-9]")

HEADER = ''''''/
"""SOAP web services generated from:
%(wsdl)s.
"""

from soaplib.serializers.primitive import (
        String, Integer, Float, Double, DateTime, Bolean, Null, Array, Map, Any
    )
from soaplib.serializers.clazz import ClassSerializer

from soaplib.service import SoapServiceBase
from soaplib.service import soapmethod
''''''

INTERFACE = ''''''/
class %(name)s(%(bases)s):
    """%(docstring)s"""
''''''

SERVICE_INTERFACE_DOCSTRING = ''''''/
SOAP service ``%(serviceName)s`` with target namespace %(tns)s.
''''''

TYPE_INTERFACE_DOCSTRING = ''''''/
SOAP %(type)s ``{%(namespace)s}%(name)s``
''''''

TYPE_MAP = ''''''/
WSDL_TYPES = {
%(items)s
}
''''''

SOAPMETHOD = ''''''    @soapmethod(%(args)s, _returns=%(response)s)''''''

METHOD = ''''''    def %(name)s(self, %(args)s):''''''

METHOD_DOCSTRING = ''''''/
        """Parameters:

        %(args)s

        Returns: %(response)s
        """
''''''

STANDARD_TYPE_NAMESPACES = [
    ''http://schemas.xmlsoap.org/soap/encoding/'',
    ''http://schemas.xmlsoap.org/wsdl/'',
    ''http://www.w3.org/2001/XMLSchema''
]

SCHEMA_TYPE_MAPPING = {
    None:                   ''%(typeName)s'',

    ''None'':                 ''None'',

    ''boolean'':              ''Boolean'',
    ''string'':               ''String'',

    ''long'':                 ''Integer'',
    ''int'':                  ''Integer'',
    ''short'':                ''Integer'',
    ''byte'':                 ''Integer'',

    ''unsignedLong'':         ''Integer'',
    ''unsignedInt'':          ''Integer'',
    ''unsignedShort'':        ''Integer'',
    ''unsignedByte'':         ''Integer'',

    ''positiveInteger'':      ''Integer'',
    ''nonPositiveInteger'':   ''Integer'',
    ''negativeInteger'':      ''Integer'',
    ''nonNegativeInteger'':   ''Integer'',

    ''float'':                ''Float'',
    ''double'':               ''Float'',

    ''decimal'':              ''Decimal'',

    ''dateTime'':             ''DateTime'',
    ''date'':                 ''DateTime'',

    ''anyURI'':               ''String'',
    ''token'':                ''String'',
    ''normalizedString'':     ''String'',

    ''base64Binary'':         ''String'',
    ''hexBinary'':            ''String'',
}

def formatDocstring(text, indent=4, colwidth=78):
    width = colwidth - indent
    joiner = ''/n'' + '' '' * indent
    return joiner.join(textwrap.wrap(text, width) + [''''])

def typeName(type, sd):
    resolved = type.resolve()
    return resolved.name or ''''

def schemaTypeName(type, sd, deps=None):

    resolved = type.resolve()
    name = resolved.name or ''''

    schemaType = SCHEMA_TYPE_MAPPING.get(name)
    if schemaType is None: # not a standard type

        # user default
        schemaType = SCHEMA_TYPE_MAPPING[None] 

        # possibly save dependency link
        if deps is not None:
            deps.append(unicode(name))

    required = type.required()
    schemaType = schemaType % dict(typeName=name, required=required)

    if type.unbounded():
        schemaType = "Array(%s)" % schemaType

    return schemaType

def normalizeIdentifier(identifier):
    if not VALID_IDENTIFIER_RE.match(identifier):
        newIdentifierLetters = []
        firstLetter = True
        for letter in identifier:
            if firstLetter:
                if VALID_IDENTIFIER_FIRST_LETTER_RE.match(letter):
                    newIdentifierLetters.append(letter)
                else:
                    newIdentifierLetters.append(''_'')
                firstLetter = False
            else:
                if VALID_IDENTIFIER_SUBSEQUENT_LETTER_RE.match(letter):
                    newIdentifierLetters.append(letter)
                else:
                    newIdentifierLetters.append(''_'')
        identifier = ''''.join(newIdentifierLetters)

    if keyword.iskeyword(identifier):
        identifier = identifier + ''_''

    return identifier

def generate(client, url=None, standardTypeNamespaces=STANDARD_TYPE_NAMESPACES, removeInputOutputMesssages=True):
    """Given a WSDL URL, return a file that could become your interfaces.py
    """

    printed = [] # sequence of type name -> string

    for sd in client.sd:

        serviceOut = StringIO()

        print >>serviceOut, HEADER % dict(
                wsdl=url,
            )

        printed.append(('''', serviceOut.getvalue(),))

        # Types

        typeMap = {}
        typeSeq = []
        typeDeps = {}
        typeAttributes = {}

        typesPrinted = []

        for type_ in sd.types:

            typeOut = StringIO()

            resolved = type_[0].resolve()
            namespaceURL = resolved.namespace()[1]
            if namespaceURL not in standardTypeNamespaces:

                if resolved.enum():
                    typeDescription = "enumeration"
                else:
                    typeDescription = "complex type"

                # Look for basess
                interfaceBases = []
                if resolved.extension():
                    def find(t):
                        for c in t.rawchildren:
                            if c.extension():
                                find(c)
                            if c.ref is not None:
                                interfaceBases.append(c.ref[0])
                    find(resolved)

                if not interfaceBases:
                    interfaceBases = [''ClassSerializer'']

                rawTypeName = typeName(type_[0], sd)

                typeInterfaceName = normalizeIdentifier(rawTypeName)

                typeMap[rawTypeName] = typeInterfaceName
                typeSeq.append((rawTypeName, typeInterfaceName,))
                typeAttributes[rawTypeName] = {}

                print >>typeOut, INTERFACE % dict(
                        name=normalizeIdentifier(typeInterfaceName),
                        bases='', ''.join(interfaceBases),
                        docstring=formatDocstring(TYPE_INTERFACE_DOCSTRING % dict(
                                type=typeDescription,
                                name=rawTypeName,
                                namespace=namespaceURL,
                            )
                        )
                    )

                print >>typeOut, "    class types:"

                if resolved.enum():
                    for attr in type_[0].children():
                        name = attr[0].name.replace('' '', ''_'')
                        print >>typeOut, "        %s = String # XXX: Enumeration value" % name
                else:
                    for attr in type_[0].children():
                        name = attr[0].name.replace('' '', ''_'')
                        attrTypeName = typeName(attr[0], sd)
                        typeAttributes[rawTypeName][name] = attrTypeName
                        schemaType = schemaTypeName(attr[0], sd, deps=typeDeps.setdefault(unicode(rawTypeName), []))
                        print >>typeOut, "        %s = %s" % (normalizeIdentifier(name), schemaType,)

                print >>typeOut

                typesPrinted.append((rawTypeName, typeOut.getvalue(),))

        serviceInterfaceOut = StringIO()

        # Main service interface
        print >>serviceInterfaceOut, INTERFACE % dict(
                name=normalizeIdentifier(sd.service.name),
                bases=u"SoapServiceBase",
                docstring=formatDocstring(SERVICE_INTERFACE_DOCSTRING % dict(
                        serviceName=sd.service.name,
                        tns=sd.wsdl.tns[1],
                    )
                )
            )

        methods = {} # name -> (response type, list of parameters,)

        for p in sd.ports:
            for m in p[1]:
                methodName = m[0]
                methodArgs = m[1]
                if methodName not in methods:
                    methodDef = p[0].method(methodName)

                    # XXX: This is discards the namespace part
                    if methodDef.soap.output.body.wrapped:

                        inputMessage  = methodDef.soap.input.body.parts[0].element[0]
                        outputMessage = methodDef.soap.output.body.parts[0].element[0]

                        if outputMessage in typeAttributes:
                            if len(typeAttributes[outputMessage]) > 0:
                                response = typeAttributes[outputMessage].values()[0]
                            else:
                                response = "None"
                        else:
                            response = outputMessage

                        # Remove types used as input/output messages
                        if removeInputOutputMesssages:
                            remove = False
                            for idx, (t, x) in enumerate(typesPrinted):
                                if t == inputMessage:
                                    remove = True
                                    break
                            if remove:
                                del typesPrinted[idx]
                                if inputMessage in typeMap:
                                    del typeMap[inputMessage]

                            remove = False
                            for idx, (t, x) in enumerate(typesPrinted):
                                if t == outputMessage:
                                    remove = True
                                    break
                            if remove:
                                del typesPrinted[idx]
                                if outputMessage in typeMap:
                                    del typeMap[outputMessage]

                    else:
                        response = methodDef.soap.output.body.parts[0].element[0]

                    methods[methodName] = (response, methodArgs,)

        for methodName in sorted(methods):

            methodArgNames = [m[0] for m in methods[methodName][1]]
            methodReturnType = methods[methodName][0]

            methodArgDetails = []
            methodArgSpecs = []

            for m in methods[methodName][1]:
                argDetail = m[1]

                # for docstring

                methodModifierParts = []

                if not argDetail.required():
                    methodModifierParts.append(''optional'')
                if argDetail.nillable:
                    methodModifierParts.append(''may be None'')

                methodModifiers = ""
                if methodModifierParts:
                    methodModifiers = '' (%s)'' % '', ''.join(methodModifierParts)

                argTypeName = typeName(argDetail, sd)

                methodSpec = "``%s`` -- %s%s" % (
                        argDetail.name,
                        argTypeName,
                        methodModifiers
                    )

                methodArgDetails.append(methodSpec)

                # for @soapmethod decorator

                schemaType = schemaTypeName(argDetail, sd)
                methodArgSpecs.append(schemaType)

            # TODO: Probably not aware of array return types
            if methodReturnType not in typeMap and methodReturnType in SCHEMA_TYPE_MAPPING:
                methodReturnType = SCHEMA_TYPE_MAPPING[methodReturnType]

            print >>serviceInterfaceOut, SOAPMETHOD % dict(
                    args='', ''.join(methodArgSpecs),
                    response=methodReturnType,
                )

            print >>serviceInterfaceOut, METHOD % dict(
                    name=normalizeIdentifier(methodName),
                    args='', ''.join(methodArgNames),
                )

            print >>serviceInterfaceOut, METHOD_DOCSTRING % dict(
                    args=''/n        ''.join(methodArgDetails),
                    response=methodReturnType,
                )

            print >>serviceInterfaceOut

        # Sort list of complex types based on internal dependencies

        def sortDeps(printed):

            printed = list(reversed(printed))

            queue = [item for item in printed if len(typeDeps.get(unicode(item[0]), [])) == 0]
            satisfied = set(queue)
            remaining = [item for item in printed if item not in queue]

            sortedPrinted = []

            while queue:
                item = queue.pop()
                itemTypeName = unicode(item[0])

                sortedPrinted.append(item)
                satisfied.add(itemTypeName)

                for item in remaining:

                    remainingItemTypeName = unicode(item[0])

                    depsList = typeDeps.get(remainingItemTypeName, [])
                    remainingDeps = []
                    for dep in depsList:
                        if dep not in satisfied:
                            remainingDeps.append(dep)

                    typeDeps[remainingItemTypeName] = remainingDeps

                    if len(remainingDeps) == 0:
                        queue.append(item)
                        remaining.remove(item)

            return sortedPrinted

        typesPrinted = sortDeps(typesPrinted)

        # Print everything
        printed.extend(typesPrinted)
        printed.append((sd.service.name, serviceInterfaceOut.getvalue(),))

        typeMapOut = StringIO()
        print >>typeMapOut, TYPE_MAP % dict(
                items='',/n''.join(["    ''%s'': %s" % k for k in typeSeq if k[0] in typeMap])
            )
        print >>typeMapOut
        printed.append(('''', typeMapOut.getvalue(),))

    return ''/n''.join([v[1] for v in printed])

def main():
    if len(sys.argv) < 2:
        print "Usage: %s <url>" % sys.argv[0]
        print "The output will be printed to the console"
        return

    if not ''://'' in sys.argv[1]:
        sys.argv[1] = ''file://'' + os.path.abspath(sys.argv[1])

    client = suds.client.Client(sys.argv[1])
    print generate(client, sys.argv[1])

if __name__ == ''__main__'':
    main()