如何装饰一个 Python 进程以捕获 HTTP 请求和响应？

Question

我目前正在调查一个非常大的 Python 代码库，它有很多副作用和意外行为，我想通过查看所有出站 HTTP 请求来了解它在做什么它在整个执行过程中，在调用堆栈中的任何一点进行。是否有任何实用程序或集成路径允许我自动分析由 Python 中编写的代码进行的完整网络调用集？

具体而言，与单独的外部工具相反，我希望能够从分析模块或相邻模块中以编程方式与捕获的 HTTP 请求和响应进行交互；例如：

使用该代码库中现有的 logging handlers 记录 requests/responses
将请求发布到类似 Kafka
解析成pandas dataframe进行分析
集成到现有 unittest or pytest 套件中

我查看了不同的可观察性工具的产品。例如，Sentry appears to automatically integrate with Python's httplib to create a "breadcrumb" for each request; however Sentry only records this information when an exception is being thrown, and its default behavior is only to publish to its Web UI. New Relic 还提供查看“外部服务”调用的功能，作为其应用程序性能监控产品的一部分，再次通过其自己的仪表板。然而，在这两种情况下，它们都缺少官方支持的 Python 处理程序，该处理程序允许上述任务在生成出站网络请求的过程中发生。

Answer 1

我查看了 Sentry's Python SDK source code，了解他们如何与 http.client 集成，并以一种概括的方式调整他们的方法以满足我的需求。

这是我编写的用于装饰 http.client.HTTPConnection 对象以利用请求、请求主体和响应对象的代码。这个特定的示例将我想要收集的数据附加到位于分析模块下的全局列表中，并将相同的数据记录到标准输出中。您可以轻松地替换任何您想要的定制功能来代替对 list.append 和 logger.info:

的调用

import logging
import sys
from http.client import HTTPConnection

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
formatter = logging.Formatter(fmt="%(name)s %(funcName)s %(levelname)s: %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)

put_request_content = []
get_response_content = []
request_bodies = []


def decorate_HTTPConnection():
    """Taken loosely from https://github.com/getsentry/sentry-python/blob/master/sentry_sdk/integrations/stdlib.py"""

    global put_request_content, get_response_content, request_bodies

    real_putrequest = HTTPConnection.putrequest
    real_getresponse = HTTPConnection.getresponse
    real__send_output = HTTPConnection._send_output

    def new_putrequest(self, method, url, skip_host=False, skip_accept_encoding=False):
        logger.info(f'{method}: {url}')
        put_request_content.append((method, url))

        real_putrequest(self, method, url, skip_host=skip_host, skip_accept_encoding=skip_accept_encoding)

    def new_getresponse(self):
        returned_response = real_getresponse(self)

        logger.info(returned_response)
        get_response_content.append(returned_response)

        return returned_response

    def new__send_output(self, message_body=None, encode_chunked=False):
        logger.info(f'Message body: {message_body}')
        request_bodies.append(message_body)

        real__send_output(self, message_body=message_body, encode_chunked=encode_chunked)

    HTTPConnection.putrequest = new_putrequest
    HTTPConnection.getresponse = new_getresponse
    HTTPConnection._send_output = new__send_output


decorate_HTTPConnection()

这是我用来测试其行为的一个非常简单的脚本：

import logging
import sys
import requests

from http_profiler.connection_decorator import put_request_content, get_response_content, request_bodies

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
formatter = logging.Formatter(fmt="%(name)s %(funcName)s %(levelname)s: %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)


def test_profile_http_get_via_requests_library(url):
    prev_len_put_request_content = len(put_request_content)
    prev_len_get_repsonse_content = len(get_response_content)
    prev_len_request_bodies = len(request_bodies)

    logger.info(f"Starting the test: GET {url}")
    resp = requests.get(url=url)

    assert resp is not None
    assert len(put_request_content) - prev_len_put_request_content == 1
    assert len(get_response_content) - prev_len_get_repsonse_content == 1
    assert len(request_bodies) - prev_len_request_bodies == 1


def test_profile_http_post_via_requests_library(url, data=None):
    if data is None:
        data = {"message": "Hello world!"}

    prev_len_put_request_content = len(put_request_content)
    prev_len_get_repsonse_content = len(get_response_content)
    prev_len_request_bodies = len(request_bodies)

    logger.info(f"Starting the test: POST {url} with {data}")
    resp = requests.post(url=url, data=data)

    assert resp is not None
    assert len(put_request_content) - prev_len_put_request_content == 1
    assert len(get_response_content) - prev_len_get_repsonse_content == 1
    assert len(request_bodies) - prev_len_request_bodies == 1


if __name__ == "__main__":
    test_profile_http_get_via_requests_library("https://example.com")
    test_profile_http_post_via_requests_library("https://example.com")
    logger.info(f'Requests: {put_request_content}')
    logger.info(f'Request bodies: {request_bodies}')
    logger.info(f'Responses: {[f"{response.status} {response.reason}" for response in get_response_content]}')

下面是测试脚本的输出：

__main__ test_profile_http_get_via_requests_library INFO: Starting the test: GET https://example.com
http_profiler.connection_decorator new_putrequest INFO: GET: /
http_profiler.connection_decorator new__send_output INFO: Message body: None
http_profiler.connection_decorator new_getresponse INFO: <http.client.HTTPResponse object at 0x7ff40aa5df10>
__main__ test_profile_http_post_via_requests_library INFO: Starting the test: POST https://example.com with {'message': 'Hello world!'}
http_profiler.connection_decorator new_putrequest INFO: POST: /
http_profiler.connection_decorator new__send_output INFO: Message body: b'message=Hello+world%21'
http_profiler.connection_decorator new_getresponse INFO: <http.client.HTTPResponse object at 0x7ff40aa5deb0>
__main__ <module> INFO: Requests: [('GET', '/'), ('POST', '/')]
__main__ <module> INFO: Request bodies: [None, b'message=Hello+world%21']
__main__ <module> INFO: Responses: ['200 OK', '200 OK']

如何装饰一个 Python 进程以捕获 HTTP 请求和响应？

How to decorate a Python process in order to capture HTTP requests and responses?

python

networking

http

python-decorators