Tornado如何记录所有API请求的响应内容？

我之前在Tornado项目里也碰到过这个需求——要完整记录请求和响应，但Tornado不像Flask、Django那样有现成的middleware直接拿响应体。不过我们可以通过重写Handler的核心方法或者用装饰器来实现，给你分享两个实用的方案：

方案一：自定义BaseHandler全局捕获（推荐全局记录场景） #

Tornado的所有请求处理都基于RequestHandler，我们可以写一个基类重写write和finish方法，把响应内容缓存下来，最后在请求结束时统一记录。

import tornado.web
import json

class BaseHandler(tornado.web.RequestHandler):
    def initialize(self):
        # 初始化容器存储响应片段
        self._response_chunks = []

    def write(self, chunk):
        # 捕获每次write输出的内容
        self._response_chunks.append(chunk)
        # 必须调用父类的write，否则响应无法正常发送
        super().write(chunk)

    def finish(self, chunk=None):
        # 如果finish带了响应内容，也要捕获
        if chunk is not None:
            self._response_chunks.append(chunk)
            super().finish(chunk)
        else:
            super().finish()
        
        # 统一记录请求和响应
        self._log_full_request_response()

    def _log_full_request_response(self):
        # 组装请求信息
        request_data = {
            "method": self.request.method,
            "uri": self.request.uri,
            "headers": dict(self.request.headers),
            "body": self.request.body.decode('utf-8') if self.request.body else ""
        }

        # 拼接响应内容，处理编码兼容（比如二进制文件）
        try:
            response_body = b''.join(self._response_chunks).decode('utf-8')
        except UnicodeDecodeError:
            # 二进制内容（如图片、文件）记录长度即可，避免乱码
            response_body = f"[Binary Content, length: {len(b''.join(self._response_chunks))}]"
        
        response_data = {
            "status_code": self.get_status(),
            "headers": dict(self._headers),
            "body": response_body
        }

        # 这里替换成你的实际日志逻辑（写入文件/日志系统等）
        print("=== 请求详情 ===")
        print(json.dumps(request_data, indent=2, ensure_ascii=False))
        print("=== 响应详情 ===")
        print(json.dumps(response_data, indent=2, ensure_ascii=False))

之后所有业务Handler只需要继承这个BaseHandler，就能自动记录每个请求的完整请求和响应了。

方案二：用装饰器实现局部记录（适合特定Handler） #

如果不需要全局记录，只想给个别Handler加日志，可以用装饰器动态替换write方法来捕获响应：

import json
from functools import wraps

def log_req_resp(func):
    @wraps(func)
    async def wrapper(self, *args, **kwargs):
        # 缓存原生write方法
        original_write = self.write
        response_chunks = []

        # 定义补丁write方法，捕获响应内容
        def patched_write(chunk):
            response_chunks.append(chunk)
            original_write(chunk)
        
        self.write = patched_write

        try:
            # 执行原Handler的业务逻辑
            await func(self, *args, **kwargs)
        finally:
            # 恢复原生write方法
            self.write = original_write
            
            # 处理并记录日志
            try:
                resp_body = b''.join(response_chunks).decode('utf-8')
            except UnicodeDecodeError:
                resp_body = f"[Binary Content, length: {len(b''.join(response_chunks))}]"
            
            request_info = {
                "method": self.request.method,
                "uri": self.request.uri,
                "body": self.request.body.decode('utf-8') if self.request.body else ""
            }
            response_info = {
                "status_code": self.get_status(),
                "body": resp_body
            }

            print("=== 请求 ===")
            print(json.dumps(request_info, indent=2))
            print("=== 响应 ===")
            print(json.dumps(response_info, indent=2))
    return wrapper

# 使用示例
class DemoHandler(tornado.web.RequestHandler):
    @log_req_resp
    async def get(self):
        self.write({"msg": "Hello Tornado"})
        self.finish()

注意事项 #

• 对于redirect这类直接发送3xx响应的方法，不会触发write，需要在finish里额外判断状态码，记录跳转地址；
• 如果是大文件下载等场景，缓存完整响应体会占用较多内存，建议只记录状态码和响应长度，跳过具体内容；
• 异步Handler（带async def）的场景，装饰器要确保用异步包装逻辑（上面的装饰器已经适配）。

内容的提问来源于stack exchange，提问作者Trần Kim Dự