def get(self, url, timeout=None):
"""Request an URL.
Args:
url (str): The web location we want to retrieve.
timeout (Optional[int|float]): If this value is specified, use it as the read timeout
from the server (instead of the one specified during creation of the connection
pool).
Returns:
A JSON object.
"""
urlopen_kwargs = {}
if timeout is not None:
urlopen_kwargs['timeout'] = Timeout(read=timeout, connect=self._connect_timeout)
result = self._request_wrapper('GET', url, **urlopen_kwargs)
return self._parse(result)
python类connection()的实例源码
def post(self, url, data, timeout=None):
"""Request an URL.
Args:
url (str): The web location we want to retrieve.
data (dict[str, str]): A dict of key/value pairs. Note: On py2.7 value is unicode.
timeout (Optional[int|float]): If this value is specified, use it as the read timeout
from the server (instead of the one specified during creation of the connection
pool).
Returns:
A JSON object.
"""
urlopen_kwargs = {}
if timeout is not None:
urlopen_kwargs['timeout'] = Timeout(read=timeout, connect=self._connect_timeout)
if InputFile.is_inputfile(data):
data = InputFile(data)
result = self._request_wrapper(
'POST', url, body=data.to_form(), headers=data.headers, **urlopen_kwargs)
else:
data = json.dumps(data)
result = self._request_wrapper(
'POST',
url,
body=data.encode(),
headers={'Content-Type': 'application/json'},
**urlopen_kwargs)
return self._parse(result)
def retrieve(self, url, timeout=None):
"""Retrieve the contents of a file by its URL.
Args:
url (str): The web location we want to retrieve.
timeout (Optional[int|float]): If this value is specified, use it as the read timeout
from the server (instead of the one specified during creation of the connection
pool).
"""
urlopen_kwargs = {}
if timeout is not None:
urlopen_kwargs['timeout'] = Timeout(read=timeout, connect=self._connect_timeout)
return self._request_wrapper('GET', url, **urlopen_kwargs)
def download(self, url, filename, timeout=None):
"""Download a file by its URL.
Args:
url (str): The web location we want to retrieve.
timeout (Optional[int|float]): If this value is specified, use it as the read timeout
from the server (instead of the one specified during creation of the connection
pool).
filename:
The filename within the path to download the file.
"""
buf = self.retrieve(url, timeout=timeout)
with open(filename, 'wb') as fobj:
fobj.write(buf)
def _request_wrapper(self, *args, **kwargs):
"""Wraps urllib3 request for handling known exceptions.
Args:
args: unnamed arguments, passed to urllib3 request.
kwargs: keyword arguments, passed tp urllib3 request.
Returns:
str: A non-parsed JSON text.
Raises:
TelegramError
"""
# Make sure to hint Telegram servers that we reuse connections by sending
# "Connection: keep-alive" in the HTTP headers.
if 'headers' not in kwargs:
kwargs['headers'] = {}
kwargs['headers']['connection'] = 'keep-alive'
try:
resp = self._con_pool.request(*args, **kwargs)
except urllib3.exceptions.TimeoutError:
raise TimedOut()
except urllib3.exceptions.HTTPError as error:
# HTTPError must come last as its the base urllib3 exception class
# TODO: do something smart here; for now just raise NetworkError
raise NetworkError('urllib3 HTTPError {0}'.format(error))
if 200 <= resp.status <= 299:
# 200-299 range are HTTP success statuses
return resp.data
try:
message = self._parse(resp.data)
except ValueError:
raise NetworkError('Unknown HTTPError {0}'.format(resp.status))
if resp.status in (401, 403):
raise Unauthorized(message)
elif resp.status == 400:
raise BadRequest(message)
elif resp.status == 404:
raise InvalidToken()
elif resp.status == 502:
raise NetworkError('Bad Gateway')
else:
raise NetworkError('{0} ({1})'.format(message, resp.status))
