Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/requests_toolbelt/utils/dump.py @ 0:26e78fe6e8c4 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:26e78fe6e8c4
1 """This module provides functions for dumping information about responses."""
2 import collections
3
4 from requests import compat
5
6
7 __all__ = ('dump_response', 'dump_all')
8
9 HTTP_VERSIONS = {
10 9: b'0.9',
11 10: b'1.0',
12 11: b'1.1',
13 }
14
15 _PrefixSettings = collections.namedtuple('PrefixSettings',
16 ['request', 'response'])
17
18
19 class PrefixSettings(_PrefixSettings):
20 def __new__(cls, request, response):
21 request = _coerce_to_bytes(request)
22 response = _coerce_to_bytes(response)
23 return super(PrefixSettings, cls).__new__(cls, request, response)
24
25
26 def _get_proxy_information(response):
27 if getattr(response.connection, 'proxy_manager', False):
28 proxy_info = {}
29 request_url = response.request.url
30 if request_url.startswith('https://'):
31 proxy_info['method'] = 'CONNECT'
32
33 proxy_info['request_path'] = request_url
34 return proxy_info
35 return None
36
37
38 def _format_header(name, value):
39 return (_coerce_to_bytes(name) + b': ' + _coerce_to_bytes(value) +
40 b'\r\n')
41
42
43 def _build_request_path(url, proxy_info):
44 uri = compat.urlparse(url)
45 proxy_url = proxy_info.get('request_path')
46 if proxy_url is not None:
47 request_path = _coerce_to_bytes(proxy_url)
48 return request_path, uri
49
50 request_path = _coerce_to_bytes(uri.path)
51 if uri.query:
52 request_path += b'?' + _coerce_to_bytes(uri.query)
53
54 return request_path, uri
55
56
57 def _dump_request_data(request, prefixes, bytearr, proxy_info=None):
58 if proxy_info is None:
59 proxy_info = {}
60
61 prefix = prefixes.request
62 method = _coerce_to_bytes(proxy_info.pop('method', request.method))
63 request_path, uri = _build_request_path(request.url, proxy_info)
64
65 # <prefix><METHOD> <request-path> HTTP/1.1
66 bytearr.extend(prefix + method + b' ' + request_path + b' HTTP/1.1\r\n')
67
68 # <prefix>Host: <request-host> OR host header specified by user
69 headers = request.headers.copy()
70 host_header = _coerce_to_bytes(headers.pop('Host', uri.netloc))
71 bytearr.extend(prefix + b'Host: ' + host_header + b'\r\n')
72
73 for name, value in headers.items():
74 bytearr.extend(prefix + _format_header(name, value))
75
76 bytearr.extend(prefix + b'\r\n')
77 if request.body:
78 if isinstance(request.body, compat.basestring):
79 bytearr.extend(prefix + _coerce_to_bytes(request.body))
80 else:
81 # In the event that the body is a file-like object, let's not try
82 # to read everything into memory.
83 bytearr.extend(b'<< Request body is not a string-like type >>')
84 bytearr.extend(b'\r\n')
85
86
87 def _dump_response_data(response, prefixes, bytearr):
88 prefix = prefixes.response
89 # Let's interact almost entirely with urllib3's response
90 raw = response.raw
91
92 # Let's convert the version int from httplib to bytes
93 version_str = HTTP_VERSIONS.get(raw.version, b'?')
94
95 # <prefix>HTTP/<version_str> <status_code> <reason>
96 bytearr.extend(prefix + b'HTTP/' + version_str + b' ' +
97 str(raw.status).encode('ascii') + b' ' +
98 _coerce_to_bytes(response.reason) + b'\r\n')
99
100 headers = raw.headers
101 for name in headers.keys():
102 for value in headers.getlist(name):
103 bytearr.extend(prefix + _format_header(name, value))
104
105 bytearr.extend(prefix + b'\r\n')
106
107 bytearr.extend(response.content)
108
109
110 def _coerce_to_bytes(data):
111 if not isinstance(data, bytes) and hasattr(data, 'encode'):
112 data = data.encode('utf-8')
113 # Don't bail out with an exception if data is None
114 return data if data is not None else b''
115
116
117 def dump_response(response, request_prefix=b'< ', response_prefix=b'> ',
118 data_array=None):
119 """Dump a single request-response cycle's information.
120
121 This will take a response object and dump only the data that requests can
122 see for that single request-response cycle.
123
124 Example::
125
126 import requests
127 from requests_toolbelt.utils import dump
128
129 resp = requests.get('https://api.github.com/users/sigmavirus24')
130 data = dump.dump_response(resp)
131 print(data.decode('utf-8'))
132
133 :param response:
134 The response to format
135 :type response: :class:`requests.Response`
136 :param request_prefix: (*optional*)
137 Bytes to prefix each line of the request data
138 :type request_prefix: :class:`bytes`
139 :param response_prefix: (*optional*)
140 Bytes to prefix each line of the response data
141 :type response_prefix: :class:`bytes`
142 :param data_array: (*optional*)
143 Bytearray to which we append the request-response cycle data
144 :type data_array: :class:`bytearray`
145 :returns: Formatted bytes of request and response information.
146 :rtype: :class:`bytearray`
147 """
148 data = data_array if data_array is not None else bytearray()
149 prefixes = PrefixSettings(request_prefix, response_prefix)
150
151 if not hasattr(response, 'request'):
152 raise ValueError('Response has no associated request')
153
154 proxy_info = _get_proxy_information(response)
155 _dump_request_data(response.request, prefixes, data,
156 proxy_info=proxy_info)
157 _dump_response_data(response, prefixes, data)
158 return data
159
160
161 def dump_all(response, request_prefix=b'< ', response_prefix=b'> '):
162 """Dump all requests and responses including redirects.
163
164 This takes the response returned by requests and will dump all
165 request-response pairs in the redirect history in order followed by the
166 final request-response.
167
168 Example::
169
170 import requests
171 from requests_toolbelt.utils import dump
172
173 resp = requests.get('https://httpbin.org/redirect/5')
174 data = dump.dump_all(resp)
175 print(data.decode('utf-8'))
176
177 :param response:
178 The response to format
179 :type response: :class:`requests.Response`
180 :param request_prefix: (*optional*)
181 Bytes to prefix each line of the request data
182 :type request_prefix: :class:`bytes`
183 :param response_prefix: (*optional*)
184 Bytes to prefix each line of the response data
185 :type response_prefix: :class:`bytes`
186 :returns: Formatted bytes of request and response information.
187 :rtype: :class:`bytearray`
188 """
189 data = bytearray()
190
191 history = list(response.history[:])
192 history.append(response)
193
194 for response in history:
195 dump_response(response, request_prefix, response_prefix, data)
196
197 return data