Locust如何将多个API调用合并为单个事务在报告/UI中展示？

当然可以实现，针对你把3个API作为一个事务、要监控每秒40个事务的需求，下面提供几种实用方案：

方案1：统一请求名称合并统计 #

直接把事务内所有API请求的name参数设为同一个事务名（比如complete_workflow），Locust会自动将这些请求的统计数据合并到同一个条目下。

修改你的代码，将每个请求的name替换为事务名：

@task
def complete_workflow(self):
    access_token = oauth.get_oauth_token(self.username, self.password)
    initiate_headers = {'Content-Type': 'application/json', "Authorization": f"Bearer {access_token}"}
    payload = {"customerInternalReference": "QEC PENN testing customer", "workflowDefinition": {"key": 3}}
    # 第一个请求指定事务名
    initiate_response = self.client.post(f"{self.account_host}/api/v1/accounts", json=payload,
                                         headers=initiate_headers, name="complete_workflow")

    # ... 中间的解析逻辑保持不变 ...

    # 第二个请求指定事务名
    self.client.post(
        f"{self.api_host}/api/v1/accounts/{self.account_id}/workflow-executions/{self.workflow_id}/credentials/{self.id_credential}/parts/FRONT",
        data=encoder, headers=headers, name="complete_workflow")

    # 第三个请求指定事务名
    finalise_header = {'Accept': '*/*', "Authorization": f"Bearer {self.api_token}"}
    self.client.put(
        f"{self.api_host}/api/v1/accounts/{self.account_id}/workflow-executions/{self.workflow_id}",
        headers=finalise_header, name="complete_workflow")

修改后，Locust的UI和报告里会把这三个请求的请求数、响应时间、成功率等数据合并到complete_workflow条目下，你可以直接查看这个条目的TPS是否达标。

方案2：用上下文管理器统计完整事务耗时（推荐） #

如果需要统计整个事务的总耗时（包括API请求之间的解析逻辑），可以使用self.client.rename_request上下文管理器，既能保留事务整体统计，也能查看单个API的细节：

@task
def complete_workflow(self):
    # 用上下文管理器包裹整个事务流程
    with self.client.rename_request("complete_workflow"):
        access_token = oauth.get_oauth_token(self.username, self.password)
        initiate_headers = {'Content-Type': 'application/json', "Authorization": f"Bearer {access_token}"}
        payload = {"customerInternalReference": "QEC PENN testing customer", "workflowDefinition": {"key": 3}}
        initiate_response = self.client.post(f"{self.account_host}/api/v1/accounts", json=payload,
                                             headers=initiate_headers)

        response = json.loads(initiate_response.text)

        workflow_credentials = response.get("workflowExecution", {}).get("credentials", [])

        id_credentials = [credential for credential in workflow_credentials if credential.get("category") == "ID"]

        selfie_credentials = [credential for credential in workflow_credentials if
                              credential.get("category") == "SELFIE"]

        self.workflow_id = response.get("workflowExecution", {}).get("id")
        self.account_id = response.get("account", {}).get("id")
        self.api_token = id_credentials[0].get("api", {}).get("token")
        self.id_credential = id_credentials[0].get("id")
        self.selfie_credential = selfie_credentials[0].get("id")

        front_image = (
            'USA_DL_front.jpg',
            open('images/USA_DL_front.jpg', 'rb'),
            "image/jpeg"
        )

        data = {'file': front_image}

        encoder = MultipartEncoder(fields=data)
        headers = {'Accept': '*/*', 'Content-Type': encoder.content_type, "Authorization": f"Bearer {self.api_token}"}

        self.client.post(
            f"{self.api_host}/api/v1/accounts/{self.account_id}/workflow-executions/{self.workflow_id}/credentials/{self.id_credential}/parts/FRONT",
            data=encoder, headers=headers)

        finalise_header = {'Accept': '*/*', "Authorization": f"Bearer {self.api_token}"}
        self.client.put(
            f"{self.api_host}/api/v1/accounts/{self.account_id}/workflow-executions/{self.workflow_id}",
            headers=finalise_header)

这种方式下，complete_workflow条目会记录整个事务的总耗时，同时每个API请求的单独统计也会保留，适合需要同时监控整体和细节的场景。

方案3：自定义独立统计项 #

如果需要完全独立的事务统计（不与单个API统计混淆），可以直接操作Locust的统计模块：

import time
from locust import stats

@task
def complete_workflow(self):
    start_time = time.time()
    success = True
    try:
        access_token = oauth.get_oauth_token(self.username, self.password)
        initiate_headers = {'Content-Type': 'application/json', "Authorization": f"Bearer {access_token}"}
        payload = {"customerInternalReference": "QEC PENN testing customer", "workflowDefinition": {"key": 3}}
        initiate_response = self.client.post(f"{self.account_host}/api/v1/accounts", json=payload,
                                             headers=initiate_headers)
        initiate_response.raise_for_status()

        # ... 中间解析逻辑 ...

        self.client.post(
            f"{self.api_host}/api/v1/accounts/{self.account_id}/workflow-executions/{self.workflow_id}/credentials/{self.id_credential}/parts/FRONT",
            data=encoder, headers=headers).raise_for_status()

        self.client.put(
            f"{self.api_host}/api/v1/accounts/{self.account_id}/workflow-executions/{self.workflow_id}",
            headers=finalise_header).raise_for_status()
    except Exception as e:
        success = False
        print(f"事务执行失败: {str(e)}")
    finally:
        elapsed_time = (time.time() - start_time) * 1000  # 转换为毫秒
        stats.stats.get_or_create_entry("complete_workflow").add_response(elapsed_time, success)

这种方式会创建一个独立的complete_workflow统计项，仅记录整个事务的执行情况，适合只关注事务整体指标的场景。

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