Create a gist now

Instantly share code, notes, and snippets.

Embed
Python xmlrpc lib Transport that used Requests
# -*- coding: utf-8 -*-
"""
A replacement transport for Python xmlrpc library.
Usage:
>>> import xmlrpclib
>>> from transport import RequestsTransport
>>> s = xmlrpclib.ServerProxy('http://yoursite.com/xmlrpc', transport=RequestsTransport())
>>> s.demo.sayHello()
Hello!
"""
try:
import xmlrpc.client as xmlrpc
except ImportError:
import xmlrpclib as xmlrpc
import requests
class RequestsTransport(xmlrpc.Transport):
"""
Drop in Transport for xmlrpclib that uses Requests instead of httplib
"""
# change our user agent to reflect Requests
user_agent = "Python XMLRPC with Requests (python-requests.org)"
# override this if you'd like to https
use_https = False
def request(self, host, handler, request_body, verbose):
"""
Make an xmlrpc request.
"""
headers = {'User-Agent': self.user_agent}
url = self._build_url(host, handler)
try:
resp = requests.post(url, data=request_body, headers=headers)
except ValueError:
raise
except Exception:
raise # something went wrong
else:
try:
resp.raise_for_status()
except requests.RequestException as e:
raise xmlrpc.ProtocolError(url, resp.status_code,
str(e), resp.headers)
else:
return self.parse_response(resp)
def parse_response(self, resp):
"""
Parse the xmlrpc response.
"""
p, u = self.getparser()
p.feed(resp.text)
p.close()
return u.close()
def _build_url(self, host, handler):
"""
Build a url for our request based on the host, handler and use_http
property
"""
scheme = 'https' if self.use_https else 'http'
return '%s://%s/%s' % (scheme, host, handler)
@petri

This comment has been minimized.

Show comment
Hide comment
@petri

petri Apr 14, 2013

To use with Plone, set the content type header to "text/xml". For Basic Auth support, get_host_info (http://hg.python.org/cpython/file/2.7/Lib/xmlrpclib.py#l1334) is your friend for getting the user & passwd to pass to requests.post.

petri commented Apr 14, 2013

To use with Plone, set the content type header to "text/xml". For Basic Auth support, get_host_info (http://hg.python.org/cpython/file/2.7/Lib/xmlrpclib.py#l1334) is your friend for getting the user & passwd to pass to requests.post.

@astraw

This comment has been minimized.

Show comment
Hide comment
@astraw

astraw May 18, 2014

@petri @chrisguitarguy the content type header is also needed with PyPI. See here for the fix.

astraw commented May 18, 2014

@petri @chrisguitarguy the content type header is also needed with PyPI. See here for the fix.

@asmeurer

This comment has been minimized.

Show comment
Hide comment
@asmeurer

asmeurer Jun 13, 2014

I don't get the point of the outer try block. Isn't

resp = requests.post(url, data=request_body, headers=headers)
try:
    resp.raise_for_status()
except requests.RequestException as e:
    raise xmlrpc.ProtocolError(url, resp.status_code, 
        str(e), resp.headers)
else:
    return self.parse_response(resp)

equivalent?

I don't get the point of the outer try block. Isn't

resp = requests.post(url, data=request_body, headers=headers)
try:
    resp.raise_for_status()
except requests.RequestException as e:
    raise xmlrpc.ProtocolError(url, resp.status_code, 
        str(e), resp.headers)
else:
    return self.parse_response(resp)

equivalent?

@TauPan

This comment has been minimized.

Show comment
Hide comment
@TauPan

TauPan Mar 22, 2018

The parse_response method is problematic, as it passes a unicode to a method that expects binary. We can simply pass the raw response object. My version (which also passes through TLS client certs) becomes:

class RequestsTransport(xmlrpclib.SafeTransport):
    """
    Drop in Transport for xmlrpclib that uses Requests instead of httplib


    """
    # change our user agent to reflect Requests
    user_agent = "Python XMLRPC with Requests (python-requests.org)"

    def __init__(self, use_https=True, cert=None, verify=None, *args, **kwargs):
        self.cert = cert
        self.verify = verify
        self.use_https = use_https
        xmlrpclib.SafeTransport.__init__(self, *args, **kwargs)

    def request(self, host, handler, request_body, verbose):
        """
        Make an xmlrpc request.
        """
        headers = {'User-Agent': self.user_agent}
        url = self._build_url(host, handler)
        try:
            resp = requests.post(url, data=request_body, headers=headers,
                                 stream=True,
                                 cert=self.cert, verify=self.verify)
        except ValueError:
            raise
        except Exception:
            raise  # something went wrong
        else:
            try:
                resp.raise_for_status()
            except requests.RequestException as e:
                raise xmlrpclib.ProtocolError(url, resp.status_code,
                                              str(e), resp.headers)
            else:
                self.verbose = verbose
                return self.parse_response(resp.raw)

    def _build_url(self, host, handler):
        """
        Build a url for our request based on the host, handler and use_http
        property
        """
        scheme = 'https' if self.use_https else 'http'
        return '%s://%s/%s' % (scheme, host, handler)

Duly note the lack (!) of a parse_response method as we simply use the one from the parent class. See also http://docs.python-requests.org/en/latest/user/quickstart/#raw-response-content

TauPan commented Mar 22, 2018

The parse_response method is problematic, as it passes a unicode to a method that expects binary. We can simply pass the raw response object. My version (which also passes through TLS client certs) becomes:

class RequestsTransport(xmlrpclib.SafeTransport):
    """
    Drop in Transport for xmlrpclib that uses Requests instead of httplib


    """
    # change our user agent to reflect Requests
    user_agent = "Python XMLRPC with Requests (python-requests.org)"

    def __init__(self, use_https=True, cert=None, verify=None, *args, **kwargs):
        self.cert = cert
        self.verify = verify
        self.use_https = use_https
        xmlrpclib.SafeTransport.__init__(self, *args, **kwargs)

    def request(self, host, handler, request_body, verbose):
        """
        Make an xmlrpc request.
        """
        headers = {'User-Agent': self.user_agent}
        url = self._build_url(host, handler)
        try:
            resp = requests.post(url, data=request_body, headers=headers,
                                 stream=True,
                                 cert=self.cert, verify=self.verify)
        except ValueError:
            raise
        except Exception:
            raise  # something went wrong
        else:
            try:
                resp.raise_for_status()
            except requests.RequestException as e:
                raise xmlrpclib.ProtocolError(url, resp.status_code,
                                              str(e), resp.headers)
            else:
                self.verbose = verbose
                return self.parse_response(resp.raw)

    def _build_url(self, host, handler):
        """
        Build a url for our request based on the host, handler and use_http
        property
        """
        scheme = 'https' if self.use_https else 'http'
        return '%s://%s/%s' % (scheme, host, handler)

Duly note the lack (!) of a parse_response method as we simply use the one from the parent class. See also http://docs.python-requests.org/en/latest/user/quickstart/#raw-response-content

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment