/usr/lib/python3/dist-packages/aioxmlrpc/client.py is in python3-aioxmlrpc 0.3-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """
XML-RPC Client with asyncio.
This module adapt the ``xmlrpc.client`` module of the standard library to
work with asyncio.
"""
import asyncio
import logging
from xmlrpc import client as xmlrpc
import aiohttp
__ALL__ = ['ServerProxy', 'Fault', 'ProtocolError']
# you don't have to import xmlrpc.client from your code
Fault = xmlrpc.Fault
ProtocolError = xmlrpc.ProtocolError
log = logging.getLogger(__name__)
class _Method:
# some magic to bind an XML-RPC method to an RPC server.
# supports "nested" methods (e.g. examples.getStateName)
def __init__(self, send, name):
self.__send = send
self.__name = name
def __getattr__(self, name):
return _Method(self.__send, "%s.%s" % (self.__name, name))
@asyncio.coroutine
def __call__(self, *args):
ret = yield from self.__send(self.__name, args)
return ret
class AioTransport(xmlrpc.Transport):
"""
``xmlrpc.Transport`` subclass for asyncio support
"""
user_agent = 'python/aioxmlrpc'
def __init__(self, use_https, *, use_datetime=False,
use_builtin_types=False, loop):
super().__init__(use_datetime, use_builtin_types)
self.use_https = use_https
self._loop = loop
self._connector = aiohttp.TCPConnector(loop=self._loop)
@asyncio.coroutine
def request(self, host, handler, request_body, verbose):
"""
Send the XML-RPC request, return the response.
This method is a coroutine.
"""
headers = {'User-Agent': self.user_agent,
#Proxy-Connection': 'Keep-Alive',
#'Content-Range': 'bytes oxy1.0/-1',
'Accept': 'text/xml',
'Content-Type': 'text/xml' }
url = self._build_url(host, handler)
response = None
try:
response = yield from aiohttp.request(
'POST', url, headers=headers, data=request_body,
connector=self._connector, loop=self._loop)
body = yield from response.text()
if response.status != 200:
raise ProtocolError(url, response.status,
body, response.headers)
except ProtocolError:
raise
except Exception as exc:
log.error('Unexpected error', exc_info=True)
raise ProtocolError(url, response.status,
str(exc), response.headers)
return self.parse_response(body)
def parse_response(self, body):
"""
Parse the xmlrpc response.
"""
p, u = self.getparser()
p.feed(body)
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)
def close(self):
self._connector.close()
class ServerProxy(xmlrpc.ServerProxy):
"""
``xmlrpc.ServerProxy`` subclass for asyncio support
"""
def __init__(self, uri, transport=None, encoding=None, verbose=False,
allow_none=False, use_datetime=False,use_builtin_types=False,
loop=None):
self._loop = loop or asyncio.get_event_loop()
if not transport:
transport = AioTransport(uri.startswith('https://'),
loop=self._loop)
super().__init__(uri, transport, encoding, verbose, allow_none,
use_datetime, use_builtin_types)
@asyncio.coroutine
def __request(self, methodname, params):
# call a method on the remote server
request = xmlrpc.dumps(params, methodname, encoding=self.__encoding,
allow_none=self.__allow_none).encode(self.__encoding)
response = yield from self.__transport.request(
self.__host,
self.__handler,
request,
verbose=self.__verbose
)
if len(response) == 1:
response = response[0]
return response
def close(self):
self.__transport.close()
def __getattr__(self, name):
return _Method(self.__request, name)
|