如果你在一个 FastAPI SQLAlchemy async pytest 的工程里跑过异步测试大概率见过下面这几条报错中的一条或几条MissingGreenlet、Event loop is closed、Future attached to a different loop。它们看起来毫不相关但追到最底层全都指向同一个根源——数据库会话AsyncSession的生命周期和 pytest-asyncio 管理的事件循环event loop生命周期没有对齐。这个问题在纯同步项目中几乎不存在因为你不会反复创建和销毁事件循环连接、事务、会话都在同一个线程模型里安安稳稳。但切到异步之后pytest-asyncio 默认的“每个测试函数一个事件循环”策略会把所有异步资源的生命周期打散数据库连接池首当其冲。这篇文章我会把这条因果链从头拆到尾然后给出一套我实际跑过很多项目、验证过可用且可复现的 conftest 配置和测试隔离方案。无论你是刚把项目迁到异步还是已经在异步测试里反复踩坑这篇文章都值得完整读完。1. 先看现象崩溃现场与报错定位1.1 最常见的三类报错长什么样我刚开始做异步测试时遇到的第一类报错是MissingGreenlet完整信息大致是sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; cant call await_only() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/14/xdb1)这个报错的直接感受是“莫名其妙”因为代码明明看着没问题。你查了半天最后发现是某个地方用了同步风格的查询或是在对象属性过期之后触发了懒加载。第二类报错是RuntimeError: Event loop is closed这个就直白多了字面意思是事件循环已经被关闭了但某个异步操作还想去调用它。通常发生在连接池归还连接、session 关闭、后台任务清理这些延迟动作上。第三类报错和 loop 的错配有关RuntimeError: Task Task pending nameTask-3 ... got Future Future pending ... attached to a different loop这种错更隐蔽意味着你创建的某个 asyncio 对象属于 loop A但真正 await 它的地方跑在 loop B 上。异步数据库连接、锁、队列、异步生成器都可能踩中。1.2 报错背后的共同元凶把这三类报错放在一起看规律就很明显了AsyncSession、连接池、事务这些对象都有明确的“归属 loop”。一旦它们跨 loop 使用或者它们所属的 loop 已经被关闭SQLAlchemy 就会用各种方式提醒你。换句话说异步数据库会话的问题从来不只是“代码写错”更多是资源生命周期和事件循环生命周期两者没对齐。下面的章节会把这个机制讲清楚。2. 为什么异步测试会和数据库会话过不去2.1 事件循环是异步世界的地基要理解这个问题先得理解事件循环event loop。异步代码不是真的并行它靠的是在一个单线程里不断切换协程而这个切换的调度中心就是事件循环。每个Task、Future、asyncio.Lock、asyncio.Queue在创建时都会和当前的事件循环绑定。你可以把事件循环想象成一个“车间”凡是这个车间里生产出来的异步工具都只能拿回这个车间用。你把它拿到别的车间去操作就会报“attached to a different loop”。SQLAlchemy 的异步引擎、连接、会话也不例外。它们底层封装了 asyncpg 或 psycopg 的异步连接这些连接本质上也是 asyncio 对象同样只能存活在特定的事件循环里。2.2 AsyncSession、连接池与事件循环的归属关系一个AsyncEngine创建时本身并不绑定某个具体 loop但它的连接池是“按 loop 管理”的。当某个协程在一个事件循环中通过 engine 获取数据库连接时这个连接会被标记为当前 loop 专属。如果之后另一个事件循环尝试使用同一个连接轻则连接直接不可用重则抛Future attached to a different loop。AsyncSession的引入让问题又多了一层。AsyncSession本身是个门面真正的数据库操作发生在底层的Connection上。当你在一个测试里创建了 session、执行了查询session 会从 engine 的连接池里取一个连接这个连接就归属当前测试的事件循环。测试结束pytest-asyncio 销毁当前事件循环但连接池并不知道“这个 loop 没了”连接还躺在池里。下一个测试再跑拿到这条“僵尸连接”自然会炸。2.3 pytest-asyncio 默认的 loop 管理策略pytest 本身是同步框架。为了让异步测试跑起来pytest-asyncio 的默认策略是每个异步测试函数都单独创建一个事件循环测试结束立刻关闭这个循环。这从隔离性上讲是合理的——每个测试互不干扰不会出现上一个测试的 Task 残留到下一个测试里。但代价也明显事件循环是“用完即弃”的。凡是生命周期跨越了测试函数边界的东西——比如 session 级 fixture 里创建的 engine、连接池、缓存下来的 session——都需要在事件循环切换后依然保持可用。而这恰恰是异步资源最不擅长的事。默认配置下只要你把 engine 或者连接池定义成 session 级就一定会遇到跨 loop 的隐患。2.4 一条典型的故障时间线我用一个具体的时序来说明这个“炸掉”的过程测试 A 开始pytest-asyncio 创建 event loop L1。测试 A 通过 session 级 fixture 拿到 engine首次执行 SQL连接池创建连接 C1C1 归属 L1。测试 A 结束L1 被关闭。测试 B 开始pytest-asyncio 创建 event loop L2。测试 B 再次通过同一个 engine 执行 SQL连接池把 C1 分配给测试 B。测试 B 使用 C1 时发现 C1 绑定的 L1 早已关闭于是抛出RuntimeError: Event loop is closed。在某些版本或组合下报错信息会变成Future attached to a different loop。本质都一样连接池跨 loop 复用了连接。3. 根因拆解四个高频诱因的真实场景3.1 诱因一fixture 作用域和 loop 作用域不一致这是最常见的原因。很多人会写一个 session 级的 engine fixture再写一个 function 级的 session fixture。代码长这样import pytest_asyncio from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker pytest_asyncio.fixture(scopesession) async def db_engine(): engine create_async_engine(postgresqlasyncpg://user:passlocalhost/test_db) yield engine await engine.dispose() pytest_asyncio.fixture async def db_session(db_engine): factory async_sessionmaker(db_engine, expire_on_commitFalse) async with factory() as session: yield session这个写法从表面上挑不出毛病但实际跑起来只要用例稍微多一点就会出现第一条Event loop is closed。原因就是db_engine是 session 级它创建于第一个测试的事件循环之后的测试虽然也在使用它但每个测试的事件循环都是新的。连接池里的连接属于老 loop新 loop 用不了。关键是很多人并不知道“session 级”只代表“fixture 代码只执行一次”并不代表“资源跨 loop 安全”。事件循环的生命周期才是真正的边界。3.2 诱因二引擎连接池跨 loop 复用就算你把 engine fixture 定义成 function 级只要 engine 使用默认连接池问题也没真正解决。SQLAlchemy 的异步引擎默认使用AsyncAdaptedQueuePool这类池化实现连接是复用的。第一次测试创建的连接可能被放回池里第二次测试又从池里取出。如果两次测试的事件循环不同这条连接就是“带病工作”。有一种临时规避办法是给异步引擎指定poolclassNullPoolengine create_async_engine(url, poolclassNullPool)NullPool不缓存连接每次请求都新建连接用完即关。这样确实能绕开“连接跨 loop 复用”的问题但代价是性能下降——每个测试都要重新建立数据库连接测试多了以后整个 suite 会明显变慢。我在一个中等规模项目里试过跑完 600 个用例用 NullPool 比连接池方案慢了将近 40%。所以它只能作为调试手段不宜作为默认方案。3.3 诱因三事务回滚方案失效异步测试里另一个高频场景是“测试数据隔离”。很多人从同步测试里带过来的习惯是测试开头开启一个事务测试结束直接回滚这样数据库里不会残留任何数据。在同步 SQLAlchemy 下这个方案非常成熟核心是一个绑定连接的外部事务。但切到异步后很多人会直接写pytest_asyncio.fixture async def db_session(db_engine): async with db_engine.connect() as conn: trans await conn.begin() session AsyncSession(bindconn, expire_on_commitFalse) yield session await trans.rollback()粗看没毛病但一旦被测代码里执行了await session.commit()问题就来了这个 commit 会直接提交掉外层事务。等你回到 fixture 的rollback()时事务早就结束了回滚的只是空气数据库里还是会留下数据。这个问题在同步时代也存在但异步场景下大家往往更关注 loop 问题反而忽略了事务边界。3.4 诱因四同步写法混入异步代码最后一个诱因非常隐秘它和事件循环本身没关系但报错时会伪装成 loop 问题。SQLAlchemy 的AsyncSession只支持await session.execute()这类异步接口。如果你不小心在某个 service 里写了session.query(User).filter(...)或者在某处触发了懒加载SQLAlchemy 会尝试在同步代码里执行 IO然后抛出MissingGreenlet。异步 session 的默认expire_on_commitTrue也是个隐藏坑。一旦事务提交对象属性全部过期下一次访问任何属性都会触发懒加载。在同步 session 下这只是性能问题在异步 session 下直接就是MissingGreenlet。所以异步项目里几乎一律要设expire_on_commitFalse。4. 可落地的解决方案一套经过验证的配置4.1 统一事件循环作用域先解决“根”的问题问题既然出在 loop 生命周期不统一最彻底的解法就是让整个测试会话只使用一个事件循环。这样 engine、连接池、session 都活在同一棵树下跨 loop 问题从根源上消失。pytest-asyncio 0.23 以上的版本支持两种全局配置方式。第一种是使用loop_scope参数import pytest_asyncio pytest_asyncio.fixture(scopesession, loop_scopesession) async def db_engine(): ...第二种更省事直接在pytest.ini里声明默认作用域[pytest] asyncio_mode auto asyncio_default_test_loop_scope session asyncio_default_fixture_loop_scope sessionasyncio_mode auto会自动识别async def test_*不需要给每个测试打pytest.mark.asyncio。asyncio_default_test_loop_scope session和asyncio_default_fixture_loop_scope session让所有测试和 fixture 共享同一个 session 级事件循环。如果你用的是较老版本的 pytest-asyncio不支持这些配置项也可以走经典路线——自己定义 session 级event_loopfixture效果等价只是会有一个弃用警告pytest.fixture(scopesession) def event_loop(): loop asyncio.new_event_loop() yield loop loop.close()我建议有条件的新项目直接采用最新版 pytest-asyncio并用loop_scope这套配置。它更显式可读性也更好。注意统一事件循环之后测试代码里不要再手动调用asyncio.run()去创建新 loop。否则你手动创建的这个 loop 和 pytest 管理的 session loop 不是同一个问题又会回来。4.2 引擎与会话 fixture 的正确写法统一了 loop 作用域之后engine 和 session 的 fixture 就可以这样写import pytest_asyncio from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.pool import NullPool from app.db.base import Base TEST_DATABASE_URL postgresqlasyncpg://postgres:postgreslocalhost:5432/test_db pytest_asyncio.fixture(scopesession, loop_scopesession) async def db_engine(): engine create_async_engine( TEST_DATABASE_URL, poolclassNullPool, echoFalse, ) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) yield engine await engine.dispose()这里有几个细节值得说create_all放在 session 级 fixture 里保证整个测试会话只要有数据库就能建表。如果你的用例会修改表结构可以在测试中手动调用drop_allcreate_all。NullPool在这里不是为规避连接复用问题而是为了简化连接生命周期管理。每个测试函数拿到的连接都是独立创建的用完即弃不会出现连接池残留。引擎一定要在yield之后await engine.dispose()否则数据库连接会一直挂着测试结束后进程可能无法正常退出。接着是 session fixturepytest_asyncio.fixture async def db_session(db_engine): connection await db_engine.connect() outer_transaction await connection.begin() session AsyncSession( bindconnection, join_transaction_modecreate_savepoint, expire_on_commitFalse, ) try: yield session finally: await session.close() await outer_transaction.rollback() await connection.close()join_transaction_modecreate_savepoint是这套方案的关键。它的作用是被测代码里执行session.commit()时并不会真正提交外层事务而是创建一个 SAVEPOINT 保存点相当于“嵌套事务”。这样 fixture 最后回滚外层事务时所有修改都会被完整撤销不会污染数据库。如果你用的是 SQLAlchemy 2.0 之前的老版本这个参数可能不存在。那时的替代方案只能要求被测代码不要 commit或者手动再做一个内层session.begin_nested()麻烦得多。所以新项目尽量用 SQLAlchemy 2.0。4.3 被测代码里通过依赖注入创建的 session 怎么处理很多 FastAPI 项目里业务代码使用的是get_db()依赖注入而不是直接使用测试 fixture。这种情况下上面的db_sessionfixture 并不会自动生效。你需要把依赖覆盖掉pytest_asyncio.fixture async def client(db_session, app): async def override_get_db(): yield db_session app.dependency_overrides[get_db] override_get_db async with AsyncClient(transportASGITransport(appapp), base_urlhttp://test) as c: yield c app.dependency_overrides.clear()这里有一个重要提醒一旦你这么覆盖依赖所有请求都会共享同一个db_session。如果你的测试是串行执行的大多数 pytest 用例都是那没问题。但如果你在单个测试函数里用asyncio.gather并发发起多个请求多个 handler 会在同一个 session 上并发执行数据库操作这时候会遇到InvalidRequestError: The AsyncSession is already asynchronously locked and used by another operation之类的问题。并发场景下我的建议是不要共享 session让依赖注入走真实逻辑也就是每次请求从 sessionmaker 里创建独立 session 去连接数据库。这样每个请求拿到独立连接并发就安全了。代价是你需要自己清理测试数据通常可以写一个“删除指定表数据”的 fixture 来兜底。4.4 测试数据隔离的完整套路综合来看我现在推荐的测试数据隔离套路分三层默认使用事务回滚用上面db_sessionfixture 的写法让测试结束后自动回滚。面对 session.commit() 也不怕靠join_transaction_modecreate_savepoint保证 commit 只提交保存点外层事务依然可以回滚。特殊场景并发、多连接单独处理并发测试不用事务回滚方案而是独立 session 显式清理。这套组合基本覆盖了日常开发的绝大多数测试场景。我从实践来看90% 的用例都能用第一层和第二层解决剩下的并发场景单独写清理逻辑即可。5. 问题排查速查表与经验沉淀5.1 报错信息对照表报错信息典型原因解决方案sqlalchemy.exc.MissingGreenlet同步风格查询混入异步代码对象属性过期后触发懒加载检查查询是否使用await session.execute()统一设置expire_on_commitFalseRuntimeError: Event loop is closed连接或 future 所属的事件循环已被关闭统一测试事件循环作用域为 session连接池不要跨 loop 复用Task ... got Future attached to a different loopasyncio 对象跨事件循环使用统一 loop 作用域不要在 fixture 里手动调用asyncio.run()sqlalchemy.exc.PendingRollbackError事务执行出错后未回滚继续使用同一个 session用async with session.begin()或捕获异常后先await session.rollback()InvalidRequestError: The AsyncSession is already asynchronously locked同一 session 被多个协程并发使用并发测试中每个任务独立 session不要共享 fixture 的 sessionsqlalchemy.exc.InvalidRequestError: Cant reconnect until invalid transaction is rolled back连接被中途丢弃事务状态未正确结束检查外层事务与 session 的关闭顺序统一使用上面的 fixture 写法这个表是我排查问题时的第一参考。遇到报错先按“是 loop 问题还是 session 问题”分类能省掉大量无效搜索时间。5.2 几个重要的实操心得第一永远不要手动在 fixture 里关闭事件循环。我知道有文章推荐下面的写法pytest.fixture(scopesession) def event_loop(): loop asyncio.new_event_loop() yield loop loop.close()这个老写法在 pytest-asyncio 0.23 之前是标配但新版本已经开始弃用这个 fixture。如果你不想每次跑测试都看到一堆 DeprecationWarning最好直接迁移到loop_scope配置。第二设置expire_on_commitFalse是异步项目的底线。默认的True在异步场景下就是定时炸弹任何一次提交后的属性访问都可能触发懒加载然后告诉你MissingGreenlet。这一条建议写进团队代码规范而不是只放在测试 fixture 里。第三连接串里的数据库必须指向独立的测试库。不管你怎么回滚、怎么清理总会有意外写库的情况发生。开发库和测试库混用是灾难性的。我见过不止一次因为测试数据污染导致线上问题的事故所以这一步不值得省。第四pytest-xdist 并行和 session 级 loop 的兼容性需要额外确认。如果用了pytest -n auto做多进程并行每个 worker 进程会有自己独立的 session 级 loop这没问题。但每个 worker 都会创建自己的引擎和连接池数据库连接数会成倍增加测试库的连接数上限要提前调大。5.3 我积累下来的一个调试技巧最后分享一个调试技巧。当你碰到无法定位的异步数据库时报错不要急着改业务代码先用最简化的方式“还原现场”pytest.mark.asyncio async def test_minimal(): engine create_async_engine(TEST_DATABASE_URL, poolclassNullPool) async with engine.connect() as conn: result await conn.execute(text(SELECT 1)) assert result.scalar() 1 await engine.dispose()这个最小测试如果通过了说明数据库和 asyncpg 本身没问题问题大概率出在你的 fixture、session 或业务调用链路上如果这个测试都过不了那就得先检查连接串、数据库服务和事件循环配置。这个“最小复现法”帮我定位过好几个看起来非常玄学的问题其实最后都是环境或配置层面的低级错误。写在最后我经历过一段被异步测试折磨的时期几乎每个用例跑完都要盯着报错猜半小时后来把所有踩过的坑串起来才发现问题的本质只有一句话事件循环的生命周期决定了一切异步资源的安全边界。在 pytest 这个框架下把 loop 作用域统一成 session再用join_transaction_modecreate_savepoint做事务回滚隔离问题就消失了大半。希望这篇把原理、报错、配置和排查过程都讲透了能帮你少走几个我走过的弯路。