我在Python中使用requests HTTP客户端库。
有时会出现HTTP请求失败的情况，我会收到一个状态码为500的HTTP响应。
这可能发生在CI或生产环境中，我会看到类似下面的内容：
AssertionError: 200 != 500 : <Response [500]>
这并没有什么帮助。
如果我可以在上面的消息中看到X-Request-ID，那就太好了。 在我的环境中，每个HTTP响应都存在此标识符。
这意味着requests库的HTTP响应对象应将其添加。
我希望repr()的输出结果看起来像 <Response [500] XejfkmxcPfhM3dqhY2HJgQAAAAM>
由于这不是我的代码，而是requests库代码，请问如何实现此功能？
code00.py:
#!/usr/bin/env python3
import sys
import requests
def __response_repr(self):
repr_headers = (
"X-Request-ID",
"Content-Encoding", # @TODO - cfati: For demo purposes only!!! DELETE (COMMENT) THIS LINE.
)
repr_parts = ["<Response [{0:d}]".format(self.status_code)]
for repr_header in repr_headers:
if repr_header in self.headers:
repr_parts.append(" {0:}".format(self.headers[repr_header]))
repr_parts.append(">")
return "".join(repr_parts)
def main(*argv):
if argv:
print("Monkey patch {0:}...\n".format(requests.models.Response))
requests.models.Response.__str__ = requests.models.Response.__repr__ # Keep the original __str__ behavior
requests.models.Response.__repr__ = __response_repr
url = "https://www.google.com"
print("Connecting to: {0:s} ...\n".format(url))
r = requests.get(url)
print("str(Response): {0:s}".format(str(r)))
print("repr(Response): {0:s}".format(repr(r)))
if __name__ == "__main__":
print("Python {0:s} {1:d}bit on {2:s}\n".format(" ".join(item.strip() for item in sys.version.split("\n")), 64 if sys.maxsize > 0x100000000 else 32, sys.platform))
main(*sys.argv[1:])
print("\nDone.")
Output:
[cfati@CFATI-5510-0:e:\Work\Dev\StackOverflow\q059193447]> "e:\Work\Dev\VEnvs\py_064_03.07.03_test0\Scripts\python.exe" code00.py
Python 3.7.3 (v3.7.3:ef4ec6ed12, Mar 25 2019, 22:22:05) [MSC v.1916 64 bit (AMD64)] 64bit on win32
Connecting to: https://www.google.com ...
str(Response): <Response [200]>
repr(Response): <Response [200]>
Done.
[cfati@CFATI-5510-0:e:\Work\Dev\StackOverflow\q059193447]> "e:\Work\Dev\VEnvs\py_064_03.07.03_test0\Scripts\python.exe" code00.py dummy_arg_to_trigger_monkey_patch
Python 3.7.3 (v3.7.3:ef4ec6ed12, Mar 25 2019, 22:22:05) [MSC v.1916 64 bit (AMD64)] 64bit on win32
Monkey patch <class 'requests.models.Response'>...
Connecting to: https://www.google.com ...
str(Response): <Response [200]>
repr(Response): <Response [200] gzip>
Done.
如果可能的话，我更愿意创建自己的“响应流程”，而不是使用任何形式的猴子补丁。我已经检查了请求库的源代码，我们可以使用请求事件钩子系统实现所需的功能。
有一个可行的示例，我们可以对响应执行任何操作。唯一的变化是我们必须为使用该功能使用自己的Session对象实例。但是！如果我们不想覆盖任何源代码，我们可以进行一行猴子补丁，将默认的Session类覆盖为默认API调用，并且它将在任何地方像魅力一样工作。
我的解决方案在这里。
import requests
class ResponseVerbose(requests.Response):
extra_header_repr = 'X-Request-Guid'
def __repr__(self):
return '<Response [{}] {}: {}>'.format(
self.status_code,
self.extra_header_repr,
self.headers.get(self.extra_header_repr, 'None')
)
class Session(requests.Session):
def __init__(self):
super().__init__()
self.hooks['response'] = self.build_response
@staticmethod
def build_response(resp, *args, **kwargs):
"""
Let's rebuild the source response into required verbose response object using all fields from the original
FYI: requests.adapters.HTTPAdapter.build_response
"""
response = ResponseVerbose()
response.status_code = resp.status_code
response.headers = resp.headers
response.encoding = resp.encoding
response.raw = resp.raw
response.reason = response.raw.reason
response.url = resp.url
response.cookies = resp.cookies.copy()
response.request = resp.request
response.connection = resp.connection
return response
def main():
url = 'https://stackoverflow.com/'
sess = Session()
print('response using our own session object: {}'.format(sess.get(url)))
import requests.api
requests.api.sessions.Session = Session
print('response using monkey patched global Session class: {}'.format(requests.get(url)))
if __name__ == '__main__':
main()
输出
# python test123.py
response using our own session object: <Response [200] X-Request-Guid: 0c446bb5-7c96-495d-a831-061f5e3c2afe>
response using monkey patched global Session class: <Response [200] X-Request-Guid: 1db5aea7-8bc9-496a-addc-1231e8543a89>
Response.__getstate__() 函数的另一个更短的示例更多信息 https://github.com/psf/requests/blob/master/requests/models.py#L654
从源代码中可以看出，如果响应内容非常大，则不应该对其进行操作，因为它会获取整个
resp.content以将响应状态转换为状态字典。因此，仅在您知道响应中没有千兆字节时才使用它 :)
这个函数看起来容易得多。
import requests
class ResponseVerbose(requests.Response):
extra_header_repr = 'X-Request-Guid'
def __repr__(self):
return '<Response [{}] {}: {}>'.format(
self.status_code,
self.extra_header_repr,
self.headers.get(self.extra_header_repr, 'None')
)
class Session(requests.Session):
def __init__(self):
super().__init__()
self.hooks['response'] = self.build_response
@staticmethod
def build_response(resp, *args, **kwargs):
"""
Let's rebuild the source response into required verbose response object using all fields from the original
FYI: requests.adapters.HTTPAdapter.build_response
"""
response = ResponseVerbose()
for k, v in resp.__getstate__().items():
setattr(response, k, v)
return response
def main():
url = 'https://stackoverflow.com/'
sess = Session()
print('response using our own session object: {}'.format(sess.get(url)))
import requests.api
requests.api.sessions.Session = Session
print('response using monkey patched global Session class: {}'.format(requests.get(url)))
if __name__ == '__main__':
main()
X-Request-Guid，仅供示例。我很容易地将此额外头配置为可配置，以展示如何以正确的方式完成此操作。build_response() 仍然很长（例如复制 cookie）。也许请求库的作者会允许一个拉取请求，使更改 Response 类更简单？ - guettli
Response覆盖__repr__来猴子补丁requests.Response。 - heemayl