FastAPI异步服务初始化兼容开发与生产模式的正确方案

开发FastAPI应用时，需要通过异步加载大型数据文件来初始化异步服务，但使用fastapi dev（开发模式）时因事件循环已运行遇到问题。以下是main.py的简化版本：

import asyncio
from fastapi import FastAPI

async def load_data(config: str):
    """
    Mock of loading external data from cloud storage

    In my scenario multiple files are loaded at once and the loaded data is cached in-memory.
    """
    await asyncio.sleep(2)  # Simulate async I/O
    return config

# Create the FastAPI app
app = FastAPI()

# Above same for all attempts

# Attempt 1: Using asyncio.run()
data = asyncio.run(load_data("Hello World!"))

@app.get("/test")
async def test():
    return {"message": data}

# Attempt 2: Using event loop directly
loop = asyncio.get_event_loop()  # also tried get_running_loop
data = loop.run_until_complete(load_data("Hello World!"))

@app.get("/test")
async def test():
    return {"message": data}

# Attempt 3: Using FastAPI's startup event
async def on_startup():
    global data
    data = await load_data("Hello World!")

    @app.get("/test")
    async def test():
        return {"message": data}

app.add_event_handler("startup", on_startup)

各尝试的问题：

• 尝试1：使用asyncio.run()在生产模式（fastapi run）正常，但开发模式报错：RuntimeError: asyncio.run() cannot be called from a running event loop
• 尝试2：直接使用事件循环loop.run_until_complete()在开发模式报错：RuntimeError: This event loop is already running
• 尝试3：使用FastAPI的startup事件在开发模式报错：RuntimeError: Cannot add middleware after an application has started.

需要实现兼容开发（fastapi dev）和生产（fastapi run）模式的异步初始化方案。

核心问题在于：

1. 开发模式下fastapi dev会提前启动事件循环，直接在全局作用域调用asyncio.run()或run_until_complete()会触发循环已运行的错误。
2. 尝试3的错误是因为在启动事件中动态添加路由，此时应用已经进入初始化流程，无法再修改路由配置。

基础实现方案（全局变量） #

使用FastAPI的startup事件完成异步初始化，且提前定义路由，避免在事件中修改应用配置：

import asyncio
from fastapi import FastAPI

# 全局变量存储初始化后的数据
data = None

async def load_data(config: str):
    """模拟从云存储加载外部数据，实际场景可同时加载多个文件并缓存到内存"""
    await asyncio.sleep(2)  # 模拟异步IO操作
    return config

app = FastAPI()

# 提前定义路由，不要在startup事件中动态添加
@app.get("/test")
async def test():
    return {"message": data}

# 定义startup事件处理函数，仅负责初始化数据
@app.on_event("startup")
async def on_startup():
    global data
    data = await load_data("Hello World!")

该方案兼容两种模式的原因：

• 生产模式：fastapi run会在应用启动前执行startup事件中的异步函数，此时事件循环尚未进入持续运行状态，异步加载可正常执行。
• 开发模式：fastapi dev虽提前启动事件循环，但startup事件是FastAPI生命周期的一部分，框架会在应用完全初始化前触发该事件，异步加载逻辑会在当前运行的事件循环中执行，不会出现循环冲突。

更优雅的实现：依赖注入 #

避免全局变量的副作用，可通过依赖注入结合app.state存储初始化数据：

import asyncio
from fastapi import FastAPI, Depends

async def load_data(config: str):
    await asyncio.sleep(2)
    return config

app = FastAPI()

# 将初始化数据存储到应用状态中
@app.on_event("startup")
async def on_startup():
    app.state.data = await load_data("Hello World!")

# 定义依赖项，用于获取初始化后的数据
async def get_data():
    return app.state.data

@app.get("/test")
async def test(data: str = Depends(get_data)):
    return {"message": data}

这种方式将数据与应用实例绑定，通过依赖注入提供给路由函数，代码更模块化，也符合FastAPI的设计理念。

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