FastAPI限制外部请求遇事件循环错误，需实现串行调用

我编写了如下FastAPI代码，尝试通过asyncio.Semaphore限制外部API的并发请求数：

from fastapi import APIRouter, Depends
import aiohttp
import asyncio

router = APIRouter()
semaphore = asyncio.Semaphore(1)

async def request_from_external_api(session):
    async with session.get("some external api endpoint") as response:
        return await response.text()

@router.post("/internal_request_1")
async def internal_request_1():
    async with semaphore:   # 限制并发请求数
        async with aiohttp.ClientSession() as session:
            return await request_from_external_api(session)
        
@router.post("/internal_request_2")
async def internal_request_2():
    async with semaphore:   # 限制并发请求数
        async with aiohttp.ClientSession() as session:
            return await request_from_external_api(session)

测试同时发起2个请求时，触发如下错误：

RuntimeError: Task <Task pending coro=<RequestResponseCycle.run_asgi() running at C:\Users\Administrator\AppData\Local\Programs\Python\Python37\lib\site-packages\uvicorn\protocols\http\httptools_impl.py:420> cb=[set.discard()]> got Future attached to a different loop

需要实现FastAPI应用逐个串行调用request_from_external_api()，解决该问题。

问题原因 #

模块初始化时创建的asyncio.Semaphore绑定了当时的事件循环，但FastAPI（Uvicorn）启动后会创建新的事件循环，导致Semaphore与后续请求任务不在同一循环中，引发跨循环错误。此外，每次请求新建aiohttp.ClientSession既不高效也可能加剧循环问题。

解决方案 #

1. 延迟创建Semaphore，绑定到运行时事件循环

通过FastAPI的lifespan上下文管理器，在应用启动时获取当前运行的事件循环，再创建Semaphore，确保其与请求任务处于同一循环。

2. 全局复用aiohttp.ClientSession

遵循aiohttp最佳实践，全局创建一个ClientSession实例，避免每次请求新建，提升性能同时避免循环相关问题。

修改后的代码 #

from fastapi import APIRouter, FastAPI
import aiohttp
import asyncio
from contextlib import asynccontextmanager

router = APIRouter()
semaphore = None
session = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global semaphore, session
    # 绑定到应用运行时的事件循环创建Semaphore
    semaphore = asyncio.Semaphore(1)
    # 初始化全局aiohttp ClientSession
    session = aiohttp.ClientSession()
    yield
    # 应用关闭时清理资源
    await session.close()

async def request_from_external_api():
    async with session.get("some external api endpoint") as response:
        return await response.text()

@router.post("/internal_request_1")
async def internal_request_1():
    async with semaphore:
        return await request_from_external_api()
        
@router.post("/internal_request_2")
async def internal_request_2():
    async with semaphore:
        return await request_from_external_api()

# 初始化应用并绑定lifespan
app = FastAPI(lifespan=lifespan)
app.include_router(router)

说明 #

• lifespan确保Semaphore和ClientSession在应用启动时创建，绑定到正确的事件循环，避免跨循环错误。
• 全局复用ClientSession减少资源开销，符合异步HTTP客户端的使用规范。
• Semaphore的async with语法确保外部API请求被串行处理，满足并发限制需求。

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