Sending MTOM requests (and receive the response)

I'll start by referring to related issues: #1066 , #781 and #599 And a related pull request: #314

I was able to solve the problem of sending a MTOM request, and I would like to share the solution here, hoping that it is useful. I'm using zeep version 4.1.0 for both non-MTOM and MTOM requests.

For me the solution was to overwrite the post_xml and post methods of zeep.transports.Transport in the following way:


def _zeep_transport_post_xml(self, address, envelope, headers):
    """
    A copy of zeep.transports.Transport.post_xml adapted to enable MTOM.
    See 'CODE_CHANGE' markings below.
    """
    # CODE_CHANGE use of `xml.etree.ElementTree.tostring` instead of `zeep.wsdl.utils.etree_to_string`
    # The latter is based on lxml.etree, and handles namespaces differently. The server didn't understand the message.
    # Also, 'xml_declaration' parameter is not included here. This would also cause the server to give an error.
    message = tostring(envelope)
    # CODE_CHANGE use the method `_zeep_transport_post`, see docs on that method
    return _zeep_transport_post(self, address, message, headers)

def _zeep_transport_post(self, address, message, headers):
    """
    A copy of zeep.transports.Transport.post adapted to enable MTOM.
    See 'CODE_CHANGE' markings below.

    The most important difference is to call the `requests.Session.post` method
    with different parameters (`files` instead of `data`) so that `requests` will
    perform a `multipart` request.
    """
    # CODE_CHANGE: added import
    from zeep.utils import get_media_type

    # CODE_CHANGE: overwrite 'Content-Type' in headers
    headers.update(
        {
            "Content-Type": 'multipart/related; type="application/xop+xml"',
        }
    )
    if self.logger.isEnabledFor(logging.DEBUG):
        log_message = message
        if isinstance(log_message, bytes):
            log_message = log_message.decode("utf-8")
        self.logger.debug("HTTP Post to %s:\n%s", address, log_message)

    # CODE_CHANGE: create 'files'
    files = [("", io.BytesIO(message))]
    response = self.session.post(
        # CODE_CHANGE: replace parameter 'data' with 'files'
        address,
        headers=headers,
        timeout=self.operation_timeout,
        files=files,
    )

    if self.logger.isEnabledFor(logging.DEBUG):
        media_type = get_media_type(response.headers.get("Content-Type", "text/xml"))

        if media_type == "multipart/related":
            log_message = response.content
        else:
            log_message = response.content
            if isinstance(log_message, bytes):
                log_message = log_message.decode(response.encoding or "utf-8")

        self.logger.debug(
            "HTTP Response from %s (status: %d):\n%s",
            address,
            response.status_code,
            log_message,
        )

    return response

I use these as follows:

from functools import partial

session = requests.Session()
...
transport = zeep.transports.Transport(session=session)
if use_mtom:
    transport.post_xml = partial(_zeep_transport_post_xml, transport)

client = Client(wsdl=..., transport=transport)

The code changes are not much, and the effect is great, it works :-) So hopefully some of the ideas can be used to create an integrated solution into zeep.

mvantellingen / python-zeep

Sending MTOM requests (and receive the response) #1344