FastAPI异步接口测试问题：Aiohttp异步失效且请求耗时过长

我想用Aiohttp测试FastAPI的异步接口，该接口通过SQLAlchemy向PostgreSQL创建对象。用JSON文件发送大量请求测试时无报错，但存在两个问题：

1. 异步功能未生效
2. 请求耗时过长

API代码 #

@router.post("/city/", response_model=city_base_Response_model)
async def create_city(city: city_create_model, request: Request):
    [requestedBy, APIKEY] = await api_honesty(request, inspect.stack()[0][3])
    city_obj = await get_city_by_name(city_name=city.city_name)
    if city_obj:
        return response_body(False, status.HTTP_409_CONFLICT, message()['OBJECT_IS_ALREADY_EXIST'], None)
    try:
        city_obj = await insert_city(city=city, requestedBy=requestedBy)
        await get_log(data=jsonable_encoder(city), requestedBy=requestedBy, fun_name=inspect.stack()[0][3], object=city_obj.id)
        return response_body(True, status.HTTP_200_OK, message()['ACTION_DONE_SUCCESSFULLY'], {**city.dict(), "id": city_obj.id})
    except Exception as exception:
        logging.error(exception)
        return response_body(False, status.HTTP_500_INTERNAL_SERVER_ERROR, message()['ACTION_FAILD'], None)

Aiohttp测试代码 #

start_time = datetime.now()

async def create(url,session, data):
    async with session.post(
        url,
        headers ={'Content-Type': 'application/json','api_key':'****'},
        json=data
        )as response:
        print("StartTime",":" + " " + str(datetime.now()))
        Status = response.status
        print(f'EndTime : {datetime.now()} ---- Status {str(response.status)}')
        return Status

async def main(urls):
    async with aiohttp.ClientSession() as session:
        async with aiofiles.open('data_create.json', 'r') as f:
            contents = await f.read()
            objects = json.loads(contents)
            for row in objects:
                results = await asyncio.gather(*[create(urls,session, row)], return_exceptions=True)
            return results

if __name__ == "__main__":
    urls="http://127.0.0.1:8000/city/"
    asyncio.run(main(urls))
    print("--- %s  requests is completed  seconds ---" % (datetime.now() - start_time))

测试日志（50次请求） #

...
StartTime : 2022-07-25 11:59:35.662056
EndTime : 2022-07-25 11:59:35.662079 ---- Status 200
StartTime : 2022-07-25 11:59:35.880552
EndTime : 2022-07-25 11:59:35.880577 ---- Status 200
--- 0:00:15.970023  requests is completed  seconds ---

1. 异步功能未生效的修复 #

你的Aiohttp测试代码逻辑错误：循环中每次仅将单个create任务传入asyncio.gather，并等待该任务完成后才进入下一次循环，本质是串行执行，完全没用到异步并发特性。

正确做法是先批量创建所有请求任务，再一次性传给asyncio.gather实现并发：

async def main(urls):
    async with aiohttp.ClientSession() as session:
        async with aiofiles.open('data_create.json', 'r') as f:
            contents = await f.read()
            objects = json.loads(contents)
            # 批量创建所有异步请求任务
            tasks = [create(urls, session, row) for row in objects]
            # 并发执行所有任务
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return results

2. 请求耗时过长的优化方案 #

检查SQLAlchemy是否为异步版本 #

API接口是async def，但如果底层get_city_by_name、insert_city等数据库操作使用同步SQLAlchemy（如直接调用session.query()而非异步的async_session.execute()），会阻塞FastAPI的事件循环，导致服务端串行处理请求，最终耗时变长。

解决方法：

• 使用SQLAlchemy异步扩展，用AsyncSession替代普通Session
• 所有数据库操作改用异步调用，如await async_session.execute(select(...))
• 配置合理的数据库连接池大小，避免连接等待

限制并发数（可选） #

一次性发起50个请求可能触发服务端限流或数据库连接池耗尽，建议用asyncio.Semaphore控制并发数：

async def create(url, session, data, semaphore):
    async with semaphore:
        async with session.post(
            url,
            headers={'Content-Type': 'application/json','api_key':'****'},
            json=data
        ) as response:
            print(f"StartTime : {datetime.now()}")
            status = response.status
            print(f'EndTime : {datetime.now()} ---- Status {status}')
            return status

async def main(urls):
    semaphore = asyncio.Semaphore(10)  # 限制同时10个请求
    async with aiohttp.ClientSession() as session:
        async with aiofiles.open('data_create.json', 'r') as f:
            contents = await f.read()
            objects = json.loads(contents)
            tasks = [create(urls, session, row, semaphore) for row in objects]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return results

检查辅助函数的异步性 #

确认api_honesty、get_log等辅助函数是否为异步实现，同步函数同样会阻塞事件循环，需改为异步版本。

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