如何计算FastAPI端点收发数据时的文件往返网络传输耗时

核心统计逻辑 #

要排除接口总响应时长里的服务端业务处理耗时，仅统计传输环节耗时，逻辑如下：

• 文件上传耗时：服务端完全接收完上传文件最后一个字节的时间 - 服务端开始接收文件第一个字节的时间
• 文件回传耗时：主机接收完下载文件最后一个字节的时间 - 服务端开始发送文件第一个字节的时间

代码实现 #

1. 上传耗时统计（服务端侧可直接完成，无需客户端配合） #

通过依赖注入的方式仅对需要统计的接口生效，避免全局中间件带来的性能损耗：

from fastapi import FastAPI, Request, UploadFile, File, Depends
import time

app = FastAPI()

async def count_upload_duration(request: Request):
    # 记录开始读取请求体（文件内容）的时间
    start_recv = time.perf_counter()
    # 完整读取请求体
    await request.body()
    # 计算上传传输耗时，存入请求上下文
    request.state.upload_cost = time.perf_counter() - start_recv

@app.post("/upload", dependencies=[Depends(count_upload_duration)])
async def upload_endpoint(request: Request, file: UploadFile = File(...)):
    # 直接获取上传传输耗时，单位为秒
    upload_cost = request.state.upload_cost
    # 此处添加你自己的文件处理逻辑
    return {"upload_transfer_duration_seconds": round(upload_cost, 4)}

2. 回传耗时统计 #

回传耗时属于端到端指标，需要客户端配合完成，以下给出两种常用统计方案：

方案一：服务端统计发送全流程耗时

仅统计服务端从开始发送到把所有字节推送到网卡的耗时，适合不需要精确端到端耗时的场景：

from fastapi.responses import StreamingResponse

@app.get("/download")
async def download_endpoint():
    # 模拟待返回的文件内容
    test_file = b"test_content" * 10000
    start_send = time.perf_counter()
    
    def file_generator():
        yield test_file
    
    resp = StreamingResponse(
        file_generator(),
        media_type="application/octet-stream"
    )
    
    # 发送完成后的回调，统计服务端侧发送耗时
    @resp.background
    def after_send_callback():
        send_cost = time.perf_counter() - start_send
        # 可自行存储该指标到日志/数据库
        print(f"服务端发送耗时: {round(send_cost,4)}s")
    
    return resp

方案二：端到端精确统计

要求客户端与服务端时钟同步（NTP同步下误差一般<10ms），通过响应头传递时间戳计算：
服务端代码修改：

@app.get("/download")
async def download_endpoint():
    test_file = b"test_content" * 10000
    # 取Unix时间戳，跨机器可对比
    start_send_ts = time.time()
    def file_generator():
        yield test_file
    return StreamingResponse(
        file_generator(),
        media_type="application/octet-stream",
        # 把发送开始时间戳放到自定义响应头
        headers={"X-Send-Start-Ts": str(start_send_ts)}
    )

客户端（Python示例）计算逻辑：

import requests
import time

resp = requests.get("http://你的服务端地址/download", stream=True)
# 读取服务端发送开始时间戳
send_start = float(resp.headers["X-Send-Start-Ts"])
# 读取完整响应内容
file_content = b""
for chunk in resp.iter_content(chunk_size=1024):
    file_content += chunk
# 计算端到端回传耗时
transfer_cost = time.time() - send_start
print(f"端到端回传耗时: {round(transfer_cost,4)}s")

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