在上下文管理器外标记Locust请求响应状态的实现方案咨询

原方案失效原因 #

Locust 中使用catch_response=True参数获取的响应对象，其success()/failure()方法的生效依赖于响应上下文管理器的作用域。当get_doc函数执行完毕返回后，该上下文已被销毁，后续再调用这两个方法不会触发Locust的请求统计逻辑，因此原方案无法生效。

兼容普通Requests的最优实现方案 #

以下方案完全满足不侵入behavior业务逻辑、支持普通Requests会话调用的约束，可根据业务复杂度选择：

方案1：手动触发统计事件（推荐，适配性更强） #

通过Locust全局统计事件上报请求结果，无需依赖响应上下文，同时对普通Requests场景做空实现兼容。

代码实现

首先导入依赖：

from locust import events
from locust.clients import HttpSession
import requests

改造get_doc函数：

def get_doc(session, meth, url):
    if isinstance(session, HttpSession):
        # Locust场景下关闭自动上报，避免统计重复
        with session.request(meth, url, catch_response=True) as resp:
            doc = parse(resp.content)
            # 保存请求元数据用于手动上报
            req_meta = {
                "request_type": meth,
                "name": url,
                "response_time": resp.elapsed.total_seconds() * 1000,
                "response_length": len(resp.content)
            }
            # 挂载状态标记方法
            doc.mark_success = lambda: events.request_success.fire(**req_meta)
            doc.mark_failure = lambda exc: events.request_failure.fire(exception=exc, **req_meta)
            return doc
    else:
        # 普通Requests场景正常执行，挂载空实现兼容调用
        resp = session.request(meth, url)
        doc = parse(resp.content)
        doc.mark_success = lambda: None
        doc.mark_failure = lambda exc: None
        return doc

业务侧调用（无任何Locust专属逻辑）

def behavior1(session):
    doc = get_doc(session, "GET", url1)
    if not doc_ok(doc):
        doc.mark_failure(Exception("Document not as expected"))
    else:
        doc.mark_success()
    doc2 = get_doc(session, "GET", url2)
    # 其余业务逻辑不变

方案2：校验逻辑入参（适合校验规则固定的场景） #

将业务校验函数作为可选参数传入get_doc，在响应上下文内完成状态标记，无需额外处理事件。

代码实现

def get_doc(session, meth, url, validator=None):
    if isinstance(session, HttpSession):
        with session.request(meth, url, catch_response=True) as resp:
            doc = parse(resp.content)
            if validator:
                is_valid, err_msg = validator(doc)
                resp.success() if is_valid else resp.failure(err_msg)
            return doc
    else:
        resp = session.request(meth, url)
        return parse(resp.content)

业务侧调用

def behavior1(session):
    # 仅传入业务校验规则，无Locust专属代码
    doc = get_doc(session, "GET", url1, validator=lambda doc: (doc_ok(doc), "Document not as expected"))
    doc2 = get_doc(session, "GET", url2, validator=lambda doc: (doc2_ok(doc), "Doc2 format error"))
    # 其余业务逻辑不变

内容的提问来源于stack exchange，提问作者yamagoya