FastAPI服务抛出‘no running event loop’异常问题咨询

错误原因 #

1. 模块加载阶段无事件循环：SINGLETON = singleton()是在模块导入时同步执行的，此时程序还未启动任何异步事件循环，但client类的__init__里直接同步创建了aiohttp.ClientSession()——而ClientSession的构造必须依赖当前活跃的事件循环来初始化内部异步资源，因此触发异常。
2. 原代码的其他小问题：
   • 异步上下文管理器需要用async with而非with
   • resp.text是异步方法，要写成resp.text()才能调用

无需懒加载的解决方法 #

核心思路是确保ClientSession在事件循环启动后再创建，可以通过异步初始化方法配合程序启动钩子实现：

修改后的代码示例 #

import asyncio
import aiohttp
from fastapi import FastAPI

class Client:
    def __init__(self, session: aiohttp.ClientSession):
        self._session = session

    async def do(self):
        async with self._session.get('http://www.google.com') as resp:
            return await resp.text()

class Singleton:
    def __init__(self):
        self._client = None

    # 异步初始化方法，确保在事件循环内执行
    async def initialize(self):
        session = aiohttp.ClientSession()
        self._client = Client(session)

    async def do(self):
        if not self._client:
            raise RuntimeError("单例未完成初始化")
        return await self._client.do()

# 定义单例实例
SINGLETON = Singleton()

# FastAPI服务中用启动钩子完成初始化
app = FastAPI()

@app.on_event("startup")
async def startup():
    await SINGLETON.initialize()

@app.get("/")
async def test_api():
    content = await SINGLETON.do()
    return {"content_preview": content[:100]}

# 本地测试用主函数
if __name__ == '__main__':
    async def main():
        await SINGLETON.initialize()
        await SINGLETON.do()
    
    asyncio.run(main())

方案说明 #

• 把ClientSession的创建逻辑移到异步的initialize方法中，避免在无事件循环的同步阶段执行。
• 在FastAPI中，通过@app.on_event("startup")钩子，在服务启动时（此时事件循环已运行）完成单例的初始化，完全不需要懒加载。
• 本地测试时，用asyncio.run包裹初始化和业务逻辑，它会自动创建并管理事件循环，确保初始化在循环内执行。

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