企业官网建设流程全解析

如果你在一个 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; can't 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 name='Task-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 一条典型的故障时间线

我用一个具体的时序来说明这个“炸掉”的过程：

1. 测试 A 开始，pytest-asyncio 创建 event loop L1。
2. 测试 A 通过 session 级 fixture 拿到 engine，首次执行 SQL，连接池创建连接 C1，C1 归属 L1。
3. 测试 A 结束，L1 被关闭。
4. 测试 B 开始，pytest-asyncio 创建 event loop L2。
5. 测试 B 再次通过同一个 engine 执行 SQL，连接池把 C1 分配给测试 B。
6. 测试 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(scope="session") async def db_engine(): engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/test_db") yield engine await engine.dispose() @pytest_asyncio.fixture async def db_session(db_engine): factory = async_sessionmaker(db_engine, expire_on_commit=False) 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这类池化实现，连接是复用的。第一次测试创建的连接可能被放回池里，第二次测试又从池里取出。如果两次测试的事件循环不同，这条连接就是“带病工作”。

有一种临时规避办法是给异步引擎指定poolclass=NullPool：

engine = create_async_engine(url, poolclass=NullPool)

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(bind=conn, expire_on_commit=False) 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_commit=True也是个隐藏坑。一旦事务提交，对象属性全部过期，下一次访问任何属性都会触发懒加载。在同步 session 下这只是性能问题，在异步 session 下直接就是MissingGreenlet。所以异步项目里几乎一律要设expire_on_commit=False。

4. 可落地的解决方案：一套经过验证的配置

4.1 统一事件循环作用域：先解决“根”的问题

问题既然出在 loop 生命周期不统一，最彻底的解法就是：让整个测试会话只使用一个事件循环。这样 engine、连接池、session 都活在同一棵树下，跨 loop 问题从根源上消失。

pytest-asyncio 0.23 以上的版本支持两种全局配置方式。第一种是使用loop_scope参数：

import pytest_asyncio @pytest_asyncio.fixture(scope="session", loop_scope="session") async def db_engine(): ...

第二种更省事，直接在pytest.ini里声明默认作用域：

[pytest] asyncio_mode = auto asyncio_default_test_loop_scope = session asyncio_default_fixture_loop_scope = session

asyncio_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(scope="session") 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 = "postgresql+asyncpg://postgres:postgres@localhost:5432/test_db" @pytest_asyncio.fixture(scope="session", loop_scope="session") async def db_engine(): engine = create_async_engine( TEST_DATABASE_URL, poolclass=NullPool, echo=False, ) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) yield engine await engine.dispose()

这里有几个细节值得说：

• create_all放在 session 级 fixture 里，保证整个测试会话只要有数据库就能建表。如果你的用例会修改表结构，可以在测试中手动调用drop_all+create_all。
• NullPool在这里不是为规避连接复用问题，而是为了简化连接生命周期管理。每个测试函数拿到的连接都是独立创建的，用完即弃，不会出现连接池残留。
• 引擎一定要在yield之后await engine.dispose()，否则数据库连接会一直挂着，测试结束后进程可能无法正常退出。

接着是 session fixture：

@pytest_asyncio.fixture async def db_session(db_engine): connection = await db_engine.connect() outer_transaction = await connection.begin() session = AsyncSession( bind=connection, join_transaction_mode="create_savepoint", expire_on_commit=False, ) try: yield session finally: await session.close() await outer_transaction.rollback() await connection.close()

join_transaction_mode="create_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(transport=ASGITransport(app=app), base_url="http://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 测试数据隔离的完整套路

综合来看，我现在推荐的测试数据隔离套路分三层：

1. 默认使用事务回滚：用上面db_sessionfixture 的写法，让测试结束后自动回滚。
2. 面对 session.commit() 也不怕：靠join_transaction_mode="create_savepoint"保证 commit 只提交保存点，外层事务依然可以回滚。
3. 特殊场景（并发、多连接）单独处理：并发测试不用事务回滚方案，而是独立 session + 显式清理。

这套组合基本覆盖了日常开发的绝大多数测试场景。我从实践来看，90% 的用例都能用第一层和第二层解决，剩下的并发场景单独写清理逻辑即可。

5. 问题排查速查表与经验沉淀

5.1 报错信息对照表

这个表是我排查问题时的第一参考。遇到报错先按“是 loop 问题还是 session 问题”分类，能省掉大量无效搜索时间。

5.2 几个重要的实操心得

第一，永远不要手动在 fixture 里关闭事件循环。我知道有文章推荐下面的写法：

@pytest.fixture(scope="session") def event_loop(): loop = asyncio.new_event_loop() yield loop loop.close()

这个老写法在 pytest-asyncio 0.23 之前是标配，但新版本已经开始弃用这个 fixture。如果你不想每次跑测试都看到一堆 DeprecationWarning，最好直接迁移到loop_scope配置。

第二，设置expire_on_commit=False是异步项目的底线。默认的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, poolclass=NullPool) 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_mode="create_savepoint"做事务回滚隔离，问题就消失了大半。希望这篇把原理、报错、配置和排查过程都讲透了，能帮你少走几个我走过的弯路。