fastapi必须通过lifespan参数管理异步数据库初始化,因模块顶层await无事件循环;需用asyncgenerator函数,在yield前连接、后断开;sqlalchemy 2.0+只需创建asyncengine并设pool_pre_ping=true,不提前connect。

FastAPI启动时用lifespan管理异步数据库初始化
直接在app = FastAPI()之后写async with或await init_db()会报错——因为此时事件循环还没跑起来,await无处可落。必须交给FastAPI的生命周期钩子,即lifespan参数。
这是唯一可靠的方式,其他“在模块顶层await”“用threading.Event阻塞主线程”等方案要么失败,要么破坏异步语义。
- 定义一个异步生成器函数,用
yield分隔启动与关闭逻辑 - 传给
FastAPI(lifespan=...),不能是普通函数,必须是AsyncGenerator - 启动部分完成数据库连接池创建、表检查、缓存预热等耗时异步操作
- 关闭部分负责
pool.close()和await pool.wait_closed(),否则可能丢连接
from contextlib import asynccontextmanager
from fastapi import FastAPI
from databases import Database
<p>database = Database("postgresql://...")</p><p>@asynccontextmanager
async def lifespan(app: FastAPI):
await database.connect() # 启动时执行
yield
await database.disconnect() # 关闭时执行</p><p>app = FastAPI(lifespan=lifespan)
</p>
SQLAlchemy 2.0+ 的AsyncEngine怎么接入lifespan?
SQLAlchemy 2.0+ 的AsyncEngine本身不提供connect()方法,它只管建连接池;真正建立首个连接要靠async with engine.begin()或await engine.connect()。但你通常不需要“立即连一次”,只需确保引擎已构造好、连接池可被后续请求使用。
更关键的是:别在lifespan里做await engine.connect(),这会白占一个连接且无意义;应把连接动作留给路由处理函数或依赖注入。
调用 Cutout.Pro 视觉处理 API 进行背景移除、人像抠图和照片增强,支持文件上传与图片 URL 输入。
- 在
lifespan中只需完成create_async_engine(...)并赋值给全局变量(如engine) - 如果需要验证DB可达性,可加一次
await engine.dialect.connect(),但生产环境慎用(增加启动延迟) - 务必设置
pool_pre_ping=True,避免连接池复用失效连接 - 不要把
sessionmaker实例挂到app.state,它不是线程/协程安全的;每次请求应新建AsyncSession
为什么不能把db对象挂到app.state再await初始化?
有人试图这样写:app.state.db = Database(...); await app.state.db.connect(),结果报RuntimeError: no running event loop。根本原因是:模块导入阶段执行代码时,uvicorn还没启动事件循环,await语法无法解析。
哪怕你用asyncio.run()包裹,也会导致事件循环嵌套、Uvicorn启动失败或连接被意外关闭。
-
app.state只是个字典容器,它不改变执行上下文的异步能力 - 所有
await调用必须发生在已运行的事件循环内,也就是lifespan或请求处理协程中 - 若强行在模块级初始化,只能用同步方式(如
psycopg2.connect),但这和FastAPI异步设计冲突,也浪费了异步IO优势
测试环境下如何跳过异步初始化?
单元测试时,你往往不想真连数据库。常见错误是重写lifespan逻辑或手动patch database.connect,但容易漏掉disconnect导致测试间污染。
推荐做法:用test_app = FastAPI(lifespan=override_lifespan),其中override_lifespan是一个空的异步生成器,或者只模拟连接对象。
- 用
pytest-asyncio确保测试协程能跑在事件循环里 - 在测试前设
app.dependency_overrides[get_db] = lambda: mock_db,比动lifespan更轻量 - 如果必须测lifespan逻辑,用
anyio.from_thread.start_blocking_portal()启动临时循环,但仅限调试,勿进CI
最易被忽略的一点:本地开发时,常把数据库URL写死在代码里,导致lifespan连的是测试库而非开发库;务必通过os.getenv("DATABASE_URL")读取,并在.env中区分环境。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!










