/usr/lib/python2.7/dist-packages/requests_toolbelt/utils/dump.py is in python-requests-toolbelt 0.7.0-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 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 | """This module provides functions for dumping information about responses."""
import collections
from requests import compat
__all__ = ('dump_response', 'dump_all')
HTTP_VERSIONS = {
9: b'0.9',
10: b'1.0',
11: b'1.1',
}
_PrefixSettings = collections.namedtuple('PrefixSettings',
['request', 'response'])
class PrefixSettings(_PrefixSettings):
def __new__(cls, request, response):
request = _coerce_to_bytes(request)
response = _coerce_to_bytes(response)
return super(PrefixSettings, cls).__new__(cls, request, response)
def _get_proxy_information(response):
if getattr(response.connection, 'proxy_manager', False):
proxy_info = {}
request_url = response.request.url
if request_url.startswith('https://'):
proxy_info['method'] = 'CONNECT'
proxy_info['request_path'] = request_url
return proxy_info
return None
def _format_header(name, value):
return (_coerce_to_bytes(name) + b': ' + _coerce_to_bytes(value) +
b'\r\n')
def _build_request_path(url, proxy_info):
uri = compat.urlparse(url)
proxy_url = proxy_info.get('request_path')
if proxy_url is not None:
return proxy_url, uri
request_path = _coerce_to_bytes(uri.path)
if uri.query:
request_path += b'?' + _coerce_to_bytes(uri.query)
return request_path, uri
def _dump_request_data(request, prefixes, bytearr, proxy_info=None):
if proxy_info is None:
proxy_info = {}
prefix = prefixes.request
method = _coerce_to_bytes(proxy_info.pop('method', request.method))
request_path, uri = _build_request_path(request.url, proxy_info)
# <prefix><METHOD> <request-path> HTTP/1.1
bytearr.extend(prefix + method + b' ' + request_path + b' HTTP/1.1\r\n')
# <prefix>Host: <request-host> OR host header specified by user
headers = request.headers.copy()
host_header = _coerce_to_bytes(headers.pop('Host', uri.netloc))
bytearr.extend(prefix + b'Host: ' + host_header + b'\r\n')
for name, value in headers.items():
bytearr.extend(prefix + _format_header(name, value))
bytearr.extend(prefix + b'\r\n')
if request.body:
if isinstance(request.body, compat.basestring):
bytearr.extend(prefix + _coerce_to_bytes(request.body))
else:
# In the event that the body is a file-like object, let's not try
# to read everything into memory.
bytearr.extend('<< Request body is not a string-like type >>')
bytearr.extend(b'\r\n')
def _dump_response_data(response, prefixes, bytearr):
prefix = prefixes.response
# Let's interact almost entirely with urllib3's response
raw = response.raw
# Let's convert the version int from httplib to bytes
version_str = HTTP_VERSIONS.get(raw.version, b'?')
# <prefix>HTTP/<version_str> <status_code> <reason>
bytearr.extend(prefix + b'HTTP/' + version_str + b' ' +
str(raw.status).encode('ascii') + b' ' +
_coerce_to_bytes(response.reason) + b'\r\n')
headers = raw.headers
for name in headers.keys():
for value in headers.getlist(name):
bytearr.extend(prefix + _format_header(name, value))
bytearr.extend(prefix + b'\r\n')
bytearr.extend(response.content)
def _coerce_to_bytes(data):
if not isinstance(data, bytes) and hasattr(data, 'encode'):
data = data.encode('utf-8')
return data
def dump_response(response, request_prefix=b'< ', response_prefix=b'> ',
data_array=None):
"""Dump a single request-response cycle's information.
This will take a response object and dump only the data that requests can
see for that single request-response cycle.
Example::
import requests
from requests_toolbelt.utils import dump
resp = requests.get('https://api.github.com/users/sigmavirus24')
data = dump.dump_response(resp)
print(data.decode('utf-8'))
:param response:
The response to format
:type response: :class:`requests.Response`
:param request_prefix: (*optional*)
Bytes to prefix each line of the request data
:type request_prefix: :class:`bytes`
:param response_prefix: (*optional*)
Bytes to prefix each line of the response data
:type response_prefix: :class:`bytes`
:param data_array: (*optional*)
Bytearray to which we append the request-response cycle data
:type data_array: :class:`bytearray`
:returns: Formatted bytes of request and response information.
:rtype: :class:`bytearray`
"""
data = data_array if data_array is not None else bytearray()
prefixes = PrefixSettings(request_prefix, response_prefix)
if not hasattr(response, 'request'):
raise ValueError('Response has no associated request')
proxy_info = _get_proxy_information(response)
_dump_request_data(response.request, prefixes, data,
proxy_info=proxy_info)
_dump_response_data(response, prefixes, data)
return data
def dump_all(response, request_prefix=b'< ', response_prefix=b'> '):
"""Dump all requests and responses including redirects.
This takes the response returned by requests and will dump all
request-response pairs in the redirect history in order followed by the
final request-response.
Example::
import requests
from requests_toolbelt.utils import dump
resp = requests.get('https://httpbin.org/redirect/5')
data = dump.dump_all(resp)
print(data.decode('utf-8'))
:param response:
The response to format
:type response: :class:`requests.Response`
:param request_prefix: (*optional*)
Bytes to prefix each line of the request data
:type request_prefix: :class:`bytes`
:param response_prefix: (*optional*)
Bytes to prefix each line of the response data
:type response_prefix: :class:`bytes`
:returns: Formatted bytes of request and response information.
:rtype: :class:`bytearray`
"""
data = bytearray()
history = list(response.history[:])
history.append(response)
for response in history:
dump_response(response, request_prefix, response_prefix, data)
return data
|