Locust中基于CSV数据发起API请求后，如何导出失败请求的请求体与响应体至CSV/Excel文件

方式一：在请求任务中直接捕获失败并记录 #

这种方式更直观，在发送请求后直接判断响应状态，若失败则手动写入请求体和响应体到文件，不需要依赖全局监听器的参数限制：

import csv
import json
from locust import HttpUser, task, events, between

# 提前初始化失败请求记录的CSV文件，设置表头
with open('failed_requests.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.writer(f, delimiter=';')
    writer.writerow(["请求类型", "接口路径", "响应时间(ms)", "响应长度", "请求体", "响应体", "错误信息"])

class MyUser(HttpUser):
    wait_time = between(1, 2)
    
    @task(1)
    def langdetect(self):
        customer = next(ssn_reader)
        # 先把请求体单独存起来，方便后续记录
        request_body = json.dumps({ "text": f"/?ssn={customer[0]}" })
        try:
            # 开启catch_response，允许我们手动标记请求失败
            r = self.client.post(
                "/api/v1/customer",
                data=request_body,
                headers=headers,
                catch_response=True
            )
            # 根据业务逻辑判断失败（比如状态码非2xx）
            if r.status_code >= 400:
                r.failure(f"HTTP错误码: {r.status_code}")
                # 追加写入失败记录
                with open('failed_requests.csv', 'a', newline='', encoding='utf-8') as f:
                    writer = csv.writer(f, delimiter=';')
                    writer.writerow([
                        "POST",
                        "/api/v1/customer",
                        round(r.elapsed.total_seconds() * 1000, 2),  # 转成毫秒并保留两位小数
                        len(r.content),
                        request_body,
                        r.text,
                        f"HTTP错误码: {r.status_code}"
                    ])
        except Exception as e:
            # 处理连接超时、DNS失败等无响应的异常情况
            with open('failed_requests.csv', 'a', newline='', encoding='utf-8') as f:
                writer = csv.writer(f, delimiter=';')
                writer.writerow([
                    "POST",
                    "/api/v1/customer",
                    "N/A",
                    "N/A",
                    request_body,
                    "N/A",
                    str(e)
                ])

@events.quitting.add_listener
def hook_quitting(environment, **kw):
    print("测试结束，已完成失败请求记录的文件写入")

方式二：扩展全局监听器，通过Locust内部属性获取数据 #

如果你更倾向于用全局事件监听的方式统一处理所有失败请求，可以利用Locust的request前置事件暂存请求体，再在request_failure事件中取出并结合响应体记录：

import csv
import json
from locust import HttpUser, task, events, between

# 初始化失败请求记录文件
failed_file = open('failed_requests.csv', 'w', encoding='utf-8')
writer = csv.writer(failed_file, delimiter=';')
writer.writerow(["请求类型", "接口路径", "响应时间(ms)", "响应长度", "请求体", "响应体", "错误信息"])

# 用字典暂存最近的请求体，用请求标识作为key（这里简单用路径+响应时间）
request_body_cache = {}

@events.request.add_listener
def hook_request(request_type, name, response_time, response_length, response, context, exception, **kw):
    # 在请求发送后，把请求体存入缓存
    if hasattr(context, 'request'):
        # 处理字节类型的请求体，转成字符串
        request_body = context.request.body.decode('utf-8') if isinstance(context.request.body, bytes) else str(context.request.body)
        cache_key = f"{name}_{response_time}"
        request_body_cache[cache_key] = request_body

@events.request_failure.add_listener
def hook_request_failure(request_type, name, response_time, response_length, exception, **kw):
    # 从缓存中取出对应的请求体
    cache_key = f"{name}_{response_time}"
    request_body = request_body_cache.get(cache_key, "无法获取")
    # 获取响应体（仅当有返回响应时存在）
    response_body = kw['response'].text if 'response' in kw and kw['response'] else "无响应"
    
    # 写入失败记录
    writer.writerow([
        request_type,
        name,
        str(response_time),
        str(response_length),
        request_body,
        response_body,
        str(exception)
    ])
    # 清理缓存，避免内存占用
    if cache_key in request_body_cache:
        del request_body_cache[cache_key]

@events.quitting.add_listener
def hook_quitting(environment, **kw):
    failed_file.close()

class MyUser(HttpUser):
    wait_time = between(1, 2)
    
    @task(1)
    def langdetect(self):
        customer = next(ssn_reader)
        r = self.client.post(
            "/api/v1/customer",
            data=json.dumps({ "text": f"/?ssn={customer[0]}" }),
            headers=headers
        )

额外提示 #

• 如果需要写入.xlsx文件，可以用openpyxl库替代csv模块，核心逻辑和上面一致，只是文件操作的API不同；
• 两种方式都考虑了无响应异常的情况（比如连接超时），避免因没有response对象导致报错；
• 方式一更适合单一任务的简单场景，方式二更适合多任务统一管理失败日志的场景。

内容的提问来源于stack exchange，提问作者Mustafa Çakar