如何在Python中对FastAPI异步接口进行压力测试_结合pytest与httpx实现

不能直接用 requests 测试 FastAPI 异步接口，因为 requests 是同步阻塞的，会卡住 asyncio event loop，导致并发请求串行执行、QPS 极低；必须使用 httpx.AsyncClient 配合 pytest-asyncio，并启用 lifespan="on" 以确保 ASGI 生命周期正常。

为什么不能直接用 requests 测试 FastAPI 异步接口

因为 requests 是同步阻塞的，它会卡住整个 asyncio event loop，导致并发请求实际变成串行执行。压测时看到的 QPS 可能只有 5–10，远低于真实能力——这不是接口慢，是测试工具拖了后腿。

必须用原生支持异步的 HTTP 客户端，httpx 是目前最稳妥的选择：它同时提供 sync/async API，与 FastAPI 的 async/await 语义完全对齐，且 pytest 支持 async test 函数（需配合 pytest-asyncio 插件）。

pytest + httpx 异步测试的最小可行配置

先确保安装正确依赖：

pip install fastapi uvicorn httpx pytest pytest-asyncio

关键点在于三处声明必须一致：

• 测试函数要加 @pytest.mark.asyncio 装饰器
• 客户端必须用 httpx.AsyncClient 实例
• FastAPI app 必须以 app=app 方式传入，不能用 base_url="http://localhost:8000" —— 否则绕过 ASGI，失去异步上下文

示例结构：

import pytest
from httpx import AsyncClient
from myapp.main import app  # 假设你的 FastAPI 实例叫 app
@pytest.mark.asyncio
async def test_health():
async with AsyncClient(app=app, base_url="https://www.ufcn.cn/link/9688c999c6508777280b6e8074ad82fa") as ac:
res = await ac.get("/health")
assert res.status_code == 200

用 pytest-asyncio 控制并发压力的关键参数

单纯写 async test 不等于压测。真正施加压力靠的是 pytest 的并发执行机制，不是单个 await ac.post(...) 的快慢。

推荐两种方式组合使用：

Bing图像创建器

必应出品基于DALL·E的AI绘图工具

• 用 pytest -n 4 启用 4 个 worker 并行跑多个测试函数（适合多 endpoint 场景）
• 在单个 test 内用 asyncio.gather() 发起批量并发请求（适合单 endpoint 高频压测）

例如模拟 100 个并发 POST 请求：

@pytest.mark.asyncio
async def test_bulk_create():
    async with AsyncClient(app=app, base_url="https://www.ufcn.cn/link/9688c999c6508777280b6e8074ad82fa") as ac:
        tasks = [ac.post("/items", json={"name": f"item-{i}"}) for i in range(100)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        # 注意：gather 不抛异常，失败结果是 Exception 实例，需手动检查
        errors = [r for r in results if isinstance(r, Exception)]
        assert len(errors) == 0

容易被忽略的 ASGI 生命周期陷阱

FastAPI 在测试模式下不会自动运行 startup/shutdown 事件——比如数据库连接池、Redis client、后台任务调度器都不会初始化。压测时可能报 AttributeError: 'NoneType' object has no attribute 'execute' 这类错，本质是依赖没就绪。

解决方案只有两个：

• 在测试前显式触发：await app.router.startup()（注意：不是 app.state 上的任意属性，而是 router 的方法）
• 更稳妥的做法是复用 Uvicorn 的测试 lifespan 逻辑，在 AsyncClient 初始化时传入 lifespan="on"（需 httpx ≥ 0.23.0）

推荐写法：

async with AsyncClient(app=app, base_url="https://www.ufcn.cn/link/9688c999c6508777280b6e8074ad82fa", lifespan="on") as ac:
    # 此时 startup 已执行，shutdown 会在 exit 时自动调用

如果用了自定义 @asynccontextmanager 的 lifespan，务必确认它不依赖真实网络或外部服务——否则测试会超时或失败。