FastAPI 里怎么用异步 SQLAlchemy？MissingGreenlet 是怎么回事？

简化版

SQLAlchemy 2.0 的异步支持是「用 greenlet 把同步的 ORM 内核桥接到 asyncio」——AsyncSession 并不是把整个 ORM 重写成了异步，而是在同步的 ORM 逻辑外面套了一层 greenlet：当内核需要发 IO 时，greenlet 会把控制权交回事件循环去 await 真正的异步驱动（asyncpg/aiomysql）。理解这一点就理解了 MissingGreenlet 报错的本质：只要在「不是由 await 触发的地方」意外触发了数据库 IO，就会抛这个错——最典型的就是懒加载（访问一个没预加载的关系属性 user.orders）和 expire_on_commit 之后再读属性。所以异步 SQLAlchemy 有三条铁律：① 关系必须预加载（selectinload/joinedload）或设成 lazy="raise" 让问题提前暴露；② AsyncSession 要配 expire_on_commit=False，否则 commit 后所有属性被标记过期、再访问就触发 IO；③ 用 2.0 风格的 select() + await session.execute()，Query 那套旧 API 在异步下不可用。会话管理上：AsyncSession 不是线程安全也不是任务安全的——每个请求一个 session（用依赖 yield），绝不能多个协程共用一个 session（会话内部有状态，并发使用会直接报错或数据错乱）。驱动要装对：PostgreSQL 用 postgresql+asyncpg://、MySQL 用 mysql+aiomysql://；注意 asyncpg 不支持 ?sslmode= 这类 libpq 参数。最后一个现实判断：如果你的应用主要瓶颈是数据库本身而不是并发连接数，同步 SQLAlchemy + def 路由（走线程池）往往更省心——异步 ORM 的复杂度是实实在在的。核心记忆：greenlet 桥接；懒加载会抛 MissingGreenlet；expire_on_commit=False；一个请求一个 session。

详细版

同步 vs 异步 SQLAlchemy 对照：

	同步	异步
引擎	create_engine	create_async_engine
会话	Session	AsyncSession
查询	session.execute(...)	await session.execute(...)
驱动	psycopg2 / pymysql	asyncpg / aiomysql
懒加载	✅ 自动	❌ 抛 MissingGreenlet
路由	def（走线程池）	async def
复杂度	低	高

# ① ★引擎与会话工厂★
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine(
    "postgresql+asyncpg://user:pwd@host/db",   # ★★驱动写在 URL 里★★
    echo=False,
    pool_size=10, max_overflow=20,
    pool_pre_ping=True,                        # ★检测失效连接★
    pool_recycle=1800,
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    ★expire_on_commit=False★,                  # ★★关键！否则 commit 后读属性会炸★★
    autoflush=False,                           # ★推荐：显式控制 flush 时机★
)

# ② ★依赖：一个请求一个 session★
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        # ★★不在这里 commit —— 事务边界在 service★★
DbDep = Annotated[AsyncSession, Depends(get_session)]

# ③ ★2.0 风格查询（★异步下只能用这套★）★
from sqlalchemy import select, update, delete, func

# 单条
user = await session.get(User, uid)                        # ★主键直取★
result = await session.execute(select(User).where(User.email == e))
user = result.scalar_one_or_none()                         # ★或 scalar_one()★

# 列表
result = await session.execute(select(User).limit(20).offset(0))
users = result.scalars().all()                             # ★★scalars() 才是对象★★

# 计数
total = await session.scalar(select(func.count()).select_from(User))

# 更新/删除（★不加载对象★）
await session.execute(update(User).where(User.id == 1).values(name="x"))
await session.execute(delete(User).where(User.id == 1))
await session.commit()

# ④ ★★预加载：异步下的必修课★★
from sqlalchemy.orm import selectinload, joinedload, contains_eager

stmt = (select(User)
        .options(selectinload(User.orders))              # ★一对多：额外一条 IN★
        .options(joinedload(User.profile))               # ★多对一：JOIN★
        .options(selectinload(User.orders).selectinload(Order.items)))  # ★嵌套★
users = (await session.execute(stmt)).scalars().unique().all()
# ★★joinedload 一对多时必须 .unique()★★

# ⑤ ★★MissingGreenlet 的典型触发★★
user = await session.get(User, 1)
print(user.orders)        # ★★MissingGreenlet：懒加载触发同步 IO★★
# ✓ 解法一：预加载
# ✓ 解法二：显式异步加载
await session.refresh(user, ["orders"])
# ✓ 解法三：用 AsyncAttrs（2.0.13+）
class Base(AsyncAttrs, DeclarativeBase): ...
orders = await user.awaitable_attrs.orders                # ★★显式 await★★

# ⑥ ★模型定义：设 lazy="raise" 提前暴露问题★
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    orders: Mapped[list["Order"]] = relationship(
        back_populates="user",
        ★lazy="raise"★,          # ★★忘了预加载时报清晰的错，而不是 MissingGreenlet★★
    )

# ⑦ ★事务★
async with session.begin():                # ★★自动 commit / 异常回滚★★
    session.add(order)
    await session.flush()                  # ★拿到自增 id★
    session.add(OrderLog(order_id=order.id))
# 或显式
try:
    ...
    await session.commit()
except Exception:
    await session.rollback()
    raise

⚠️ 三个必须记住的点：① MissingGreenlet 的本质是「在非 await 的上下文里触发了数据库 IO」。SQLAlchemy 的异步支持是用 greenlet 把同步的 ORM 内核包起来——所有 IO 必须经过 await 入口（await session.execute()、await session.get()）才能正确地把控制权交给事件循环。而懒加载是在「属性访问」这个纯同步的动作里发起 IO 的，此时没有 greenlet 上下文，就抛出 MissingGreenlet: greenlet_spawn has not been called。同类的触发点还有：commit 后访问被 expire 的属性、在 __repr__ 里访问关系、Pydantic 序列化时访问未加载的关系（from_attributes=True 会遍历字段）。② expire_on_commit=False 几乎是必配项。SQLAlchemy 默认在 commit() 后把所有实例标记为「过期」，下次访问任何属性都会重新查数据库刷新——同步下这只是多一次查询，异步下就是一个不经 await 的 IO，直接抛 MissingGreenlet。典型场景：service 里 await session.commit() 之后 return order，路由把它交给 response_model 序列化时炸掉。③ AsyncSession 不是并发安全的。它内部维护着 identity map、待刷新队列、事务状态——多个协程同时用同一个 session 会直接报错（InvalidRequestError）或产生数据错乱。所以 asyncio.gather 里的多个查询不能共用一个 session，要么串行执行，要么各自开一个 session。

完整版教学

一、greenlet 桥接机制

★ ★为什么需要 greenlet★：
  SQLAlchemy 的 ORM 内核是★十几年积累的同步代码★
  （identity map、工作单元、关系加载、方言处理…）
  ★完全重写成 async 不现实★
  → 方案：★用 greenlet 做"同步代码 ↔ 异步 IO"的转换层★

★ ★执行流程★：
  ┌────────────────────────────────────────────────────────┐
  │ await session.execute(stmt)                              │
  │   ↓ ★greenlet_spawn：进入一个 greenlet★                   │
  │ ★同步的 ORM 内核开始工作★（编译 SQL、准备参数…）           │
  │   ↓ 需要真正发送 SQL 了                                   │
  │ ★await_only()：把控制权"抛"回事件循环★                     │
  │   ↓                                                      │
  │ ★事件循环 await asyncpg 的真异步 IO★                      │
  │   ↓ IO 完成                                              │
  │ ★恢复 greenlet，同步内核继续处理结果★                      │
  │   ↓                                                      │
  │ 返回 Result                                              │
  └────────────────────────────────────────────────────────┘

★ ★关键约束：IO 必须发生在 greenlet 上下文里★
  ✓ await session.execute(...)   → ★有 greenlet★
  ✓ await session.get(...)       → ★有 greenlet★
  ✗ user.orders（属性访问）      → ★★没有 greenlet → MissingGreenlet★★
  ✗ commit 后访问过期属性        → ★同上★

★ ★报错原文与含义★：
  sqlalchemy.exc.MissingGreenlet:
  greenlet_spawn has not been called; can't call await_only() here.
  Was IO attempted in an unexpected place?
  ★ 翻译：★"你在一个我没准备好做 IO 的地方尝试了 IO"★
  ★ ★最后一句 "unexpected place" 就是提示：找那个隐式的 IO★

★ ★怎么定位是哪一行★：
  ① ★看 traceback 里最后一个你的代码帧★
  ② ★常见嫌疑：★
     - ★序列化时（Pydantic model_validate / FastAPI response_model）★
     - ★模板渲染时★
     - ★日志里 f"{obj}"（触发 __repr__）★
     - ★列表推导里访问关系★
  ③ ★终极手段：给所有 relationship 设 lazy="raise"★
     → ★报错变成清晰的 "lazy load operation of attribute 'orders'
       cannot proceed"，并且指出是哪个属性★

★ ★AsyncAttrs：官方给的"显式懒加载"★（2.0.13+）
  class Base(AsyncAttrs, DeclarativeBase): pass
  orders = await user.★awaitable_attrs★.orders
  ★ ✓ 需要时能按需加载，语义明确
  ★ ✗ ★仍然是 N+1★（循环里用就是灾难）
  → ★只适合"单个对象、确实需要按需加载"的场景★

理解 greenlet 桥接就理解了一切：SQLAlchemy 的 ORM 内核是十几年的同步代码，完全重写成 async 不现实，所以用 greenlet 做转换层——await session.execute() 会 greenlet_spawn 进入一个 greenlet，同步内核在里面工作，需要真正发 SQL 时把控制权「抛」回事件循环去 await 异步驱动。关键约束是「IO 必须发生在 greenlet 上下文里」——而属性访问（懒加载）是纯同步动作，没有 greenlet 上下文，所以抛 MissingGreenlet。报错原文的最后一句 「Was IO attempted in an unexpected place?」就是提示。定位方法：看 traceback 最后一个自己的代码帧，常见嫌疑是序列化时、日志里的 f-string（触发 __repr__）、列表推导里访问关系；终极手段是给所有 relationship 设 lazy="raise"，报错会变成清晰的「哪个属性不能懒加载」。

二、预加载策略

★ ★三种加载策略★：
  ┌──────────────────┬────────────────────────────────────┐
  │ ★selectinload★    │ ★额外一条 WHERE id IN (...)★        │
  │                   │ ✓ ★一对多/多对多首选★               │
  │                   │ ✓ ★不产生笛卡尔积★                  │
  │                   │ ✓ 可嵌套                            │
  │ ★joinedload★      │ ★LEFT OUTER JOIN，一条 SQL★         │
  │                   │ ✓ ★多对一/一对一首选★               │
  │                   │ ✗ ★一对多会行数放大 + 必须 .unique()★│
  │ ★contains_eager★  │ ★配合手写 JOIN 用★                  │
  │                   │ ✓ 需要在 JOIN 上加条件时            │
  └──────────────────┴────────────────────────────────────┘

★ ★算例：100 个用户，每人 10 个订单★
  ┌────────────────────┬────────┬──────────────────┐
  │ 懒加载（同步）      │ 101 条 │ ★异步下直接报错★ │
  │ ★joinedload★        │ 1 条   │ ★返回 1000 行★   │
  │ ★selectinload★      │ ★2 条★ │ ★100 + 1000 行★  │
  │                     │        │ ★无重复★         │
  └────────────────────┴────────┴──────────────────┘
  ★ ★selectinload 的两条 SQL 通常比 joinedload 的一条大 JOIN 更快★
    （★数据传输量小、Python 侧不用去重★）

★ ★嵌套预加载★：
  stmt = select(User).options(
      selectinload(User.orders).selectinload(Order.items),   # ★两层★
      selectinload(User.orders).joinedload(Order.coupon),    # ★分支★
      joinedload(User.profile),
  )
  ★ ★注意：链式写法表示"路径"，不是并列★
    selectinload(A.b).selectinload(B.c)  → ★A → b → c★

★ ★joinedload 一对多必须 unique()★：
  users = (await session.execute(
      select(User).options(joinedload(User.orders))
  )).scalars().★unique()★.all()
  ★ ✗ 不写 unique() → ★InvalidRequestError: The unique() method
    must be invoked on this Result★
  ★ 原因：JOIN 让同一个 User 出现多行，需要去重

★ ★contains_eager：JOIN 上加条件★：
  stmt = (select(User)
          .join(User.orders)
          .where(Order.status == "paid")           # ★★过滤关联表★★
          .options(contains_eager(User.orders)))
  ★ ✓ 只加载"已支付的订单"到 user.orders
  ★ ✗ 注意：★user.orders 里就只有已支付的了★（语义上是"部分加载"）

★ ★raiseload：防止意外懒加载★：
  # ① 模型级（★推荐★）
  orders: Mapped[list["Order"]] = relationship(★lazy="raise"★)
  # ② 查询级：把没显式加载的都禁掉
  from sqlalchemy.orm import raiseload
  select(User).options(raiseload("*"))
  ★ ★价值：把"运行时 MissingGreenlet"变成"开发期就能发现的明确报错"★
  ★ 需要时用 lazy="raise_on_sql"（★已加载的可以访问，未加载的才报错★）

★ ★只查需要的列（避免加载大字段）★：
  from sqlalchemy.orm import load_only, defer
  select(User).options(load_only(User.id, User.name))
  select(Article).options(defer(Article.content))    # ★延迟大文本★
  # 或直接查列（★不构造 ORM 对象，最快★）
  rows = (await session.execute(
      select(User.id, User.name).limit(100))).all()   # ★→ Row 元组★

三种加载策略要按关系方向选：selectinload 是一对多/多对多的首选（额外一条 IN 查询，不产生笛卡尔积、可嵌套）；joinedload 是多对一/一对一的首选（一条 SQL，但一对多会行数放大且必须 .unique()，否则报 InvalidRequestError）；contains_eager 用于「需要在 JOIN 上加条件」（比如只加载已支付的订单，注意这时 user.orders 里就只有部分数据了）。算例很直观：100 个用户各 10 个订单，joinedload 一条 SQL 返回 1000 行、selectinload 两条 SQL 共 1100 行但无重复——后者通常更快（数据传输量小、Python 侧不用去重）。最有价值的实践是 lazy="raise"——它把「运行时的 MissingGreenlet」变成「开发期就能发现的明确报错」。

三、会话管理与并发

★ ★AsyncSession 不是并发安全的★：
  ✗ 危险：
    async def handler(session: AsyncSession):
        await asyncio.gather(
            session.execute(stmt1),        # ★★同一个 session★★
            session.execute(stmt2),
        )
    → ★InvalidRequestError: This session is provisioning a new
      connection; concurrent operations are not permitted★
  ✓ 方案一：★串行★（大多数场景够用）
    r1 = await session.execute(stmt1)
    r2 = await session.execute(stmt2)
  ✓ 方案二：★各自开 session★
    async def q(stmt):
        async with AsyncSessionLocal() as s:
            return (await s.execute(stmt)).scalars().all()
    r1, r2 = await asyncio.gather(q(stmt1), q(stmt2))
    ★ 注意：★这样它们不在同一事务里★

★ ★为什么不安全★：
  Session 内部有：
    - ★identity map（对象缓存）★
    - ★待 flush 的变更队列★
    - ★当前事务状态、连接绑定★
  → ★两个协程交错操作会破坏这些状态★

★ ★scoped_session 在异步下★：
  from sqlalchemy.ext.asyncio import async_scoped_session
  AsyncScopedSession = async_scoped_session(
      AsyncSessionLocal, scopefunc=★asyncio.current_task★)
  ★ ✓ 按 task 隔离（类似线程本地）
  ★ ✗ ★FastAPI 里通常不需要★——依赖注入已经保证了一请求一 session
  ★ ★而且忘了 remove() 会泄漏★

★ ★事务边界的正确位置★：
  ✗ 在依赖里 commit：
    async def get_session():
        async with AsyncSessionLocal() as s:
            yield s
            await s.commit()          # ★★service 失去事务控制权★★
  ✓ ★service 层控制★：
    class OrderService:
        async def create(self, data) -> Order:
            async with self.session.begin():      # ★★显式事务★★
                order = Order(**data.model_dump())
                self.session.add(order)
                await self.session.flush()        # ★拿 id★
                await self._deduct_stock(order)
                # ★退出 with 时自动 commit，异常时自动 rollback★
            return order

★ ★flush vs commit★：
  await session.flush()     # ★发 SQL 但★不提交★，能拿到自增 id★
  await session.commit()    # ★提交事务★
  ★ 典型用法：★一个事务里先 flush 拿 id，再插关联记录，最后统一 commit★

★ ★连接池的算术（★容易爆★）★：
  uvicorn workers=4 × pool_size=10 + max_overflow=20
  = ★4 × 30 = 120 个连接★
  PostgreSQL 默认 max_connections = 100  → ★★连不上★★
  ✓ ★调小 pool_size★（异步下每个连接利用率高，5~10 通常够）
  ✓ ★上 PgBouncer★（★注意：transaction 模式下要禁用 prepared statement 缓存★）
    create_async_engine(url, connect_args={"statement_cache_size": 0})

★ ★asyncpg 的注意事项★：
  ① ★不支持 libpq 的 URL 参数★：
     ✗ postgresql+asyncpg://...?sslmode=require
     ✓ create_async_engine(url, connect_args={"ssl": "require"})
  ② ★prepared statement 缓存与 PgBouncer 冲突★（见上）
  ③ ★类型处理更严格★（比如 JSON 要显式 codec）
  ④ ★不支持多语句★（一次 execute 一条）

AsyncSession 不是并发安全的——它内部有 identity map、待 flush 队列、事务状态，两个协程交错操作会破坏这些状态，asyncio.gather 里共用一个 session 会直接报 InvalidRequestError；要并发查询就各自开 session（代价是不在同一事务里）。事务边界应该在 service 层而不是依赖里——依赖里自动 commit 会让 service 失去事务控制权；推荐用 async with session.begin()（自动 commit、异常自动 rollback）。flush 和 commit 的区别要清楚：flush 发 SQL 但不提交、能拿到自增 id，典型用法是「先 flush 拿 id 再插关联记录，最后统一 commit」。连接池的算术容易爆：4 个 worker × (pool 10 + overflow 20) = 120，超过 PG 默认的 100。asyncpg 有四个坑：不支持 ?sslmode= 这类 libpq 参数、PgBouncer transaction 模式下要设 statement_cache_size=0、类型处理更严格、不支持多语句。

四、与 FastAPI 的集成

★ ★完整的依赖链★：
  # db/session.py
  engine = create_async_engine(settings.async_database_url, ...)
  AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

  async def get_session() -> AsyncGenerator[AsyncSession, None]:
      async with AsyncSessionLocal() as session:
          try:
              yield session
          except Exception:
              await session.rollback()
              raise
  # api/deps.py
  DbDep = Annotated[AsyncSession, Depends(get_session)]

  # service 注入
  async def get_user_service(db: DbDep) -> UserService:
      return UserService(db)
  UserServiceDep = Annotated[UserService, Depends(get_user_service)]

★ ★★response_model 与懒加载的冲突（高频事故）★★：
  @router.get("/users/{uid}", response_model=UserOut)
  async def get_user(uid: int, db: DbDep):
      return await db.get(User, uid)          # ★没预加载 orders★

  class UserOut(BaseModel):
      model_config = ConfigDict(from_attributes=True)
      id: int
      orders: list[OrderOut] = []             # ★★序列化时访问 → MissingGreenlet★★
  ✓ 解法：
    ① ★查询时预加载★（最正确）
    ② ★UserOut 里不要这个字段★（另开一个接口）
    ③ 用 AsyncAttrs 显式 await（单个对象时）

★ ★lifespan 里初始化与释放★：
  @asynccontextmanager
  async def lifespan(app: FastAPI):
      yield
      await engine.dispose()                  # ★★优雅关闭连接池★★
  ★ ✗ 不 dispose 会导致：★关闭时 "Event loop is closed" 警告★

  ★ ✗ ★不要在 lifespan 里 create_all()（生产）★
    → ★用 Alembic 迁移★
  ★ ✓ 测试和本地开发可以：
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)   # ★★run_sync★★

★ ★run_sync：调用同步的 SQLAlchemy API★：
  async with engine.begin() as conn:
      await conn.run_sync(Base.metadata.create_all)
      tables = await conn.run_sync(
          lambda sync_conn: inspect(sync_conn).get_table_names())
  ★ ★用途：建表、反射、Alembic 的同步 API★

★ ★Alembic 的异步配置★：
  # env.py
  async def run_migrations_online():
      connectable = create_async_engine(url)
      async with connectable.connect() as conn:
          await conn.run_sync(do_run_migrations)     # ★★run_sync★★
      await connectable.dispose()
  asyncio.run(run_migrations_online())
  ★ ★或者：迁移用同步驱动，运行时用异步驱动★（★更简单★）
    alembic.ini 里用 postgresql://（psycopg2）
    应用里用 postgresql+asyncpg://

★ ★分页的实现★：
  async def paginate(session, stmt, page: int, size: int):
      total = await session.scalar(
          select(func.count()).select_from(stmt.subquery()))   # ★★COUNT★★
      items = (await session.execute(
          stmt.limit(size).offset((page - 1) * size))).scalars().all()
      return {"items": items, "total": total,
              "pages": (total + size - 1) // size}
  ★ ★大表上 COUNT 很贵★ → 估算或用游标分页

★ ★测试配置★：
  @pytest_asyncio.fixture(★loop_scope="session"★)
  async def async_engine():
      eng = create_async_engine("postgresql+asyncpg://.../test")
      async with eng.begin() as conn:
          await conn.run_sync(Base.metadata.create_all)
      yield eng
      await eng.dispose()

  @pytest_asyncio.fixture
  async def session(async_engine):
      async with async_engine.connect() as conn:
          trans = await conn.begin()
          async with AsyncSession(bind=conn,
                                  expire_on_commit=False) as s:
              yield s
          await trans.rollback()          # ★★回滚做隔离★★

FastAPI 集成里最高频的事故是 response_model 与懒加载的冲突——查询时没预加载 orders，但 UserOut 里声明了这个字段，Pydantic 序列化时遍历字段会触发懒加载然后 MissingGreenlet；解法是查询时预加载、或者输出模型里不放这个字段。lifespan 里要 await engine.dispose()（否则关闭时会有「Event loop is closed」警告），但不要在生产的 lifespan 里 create_all()——用 Alembic 迁移。run_sync 是调用同步 SQLAlchemy API 的桥梁（建表、反射、Alembic）。Alembic 的异步配置比较麻烦，更简单的做法是「迁移用同步驱动、运行时用异步驱动」。测试要注意 loop_scope 对齐并用事务回滚做隔离。

五、性能与常见错误

★ ★异步 ORM 真的更快吗（★诚实的回答★）★：
  ┌────────────────────────────────────────────────────┐
  │ ★单条查询的延迟★：★几乎没区别★（瓶颈在数据库）        │
  │ ★高并发下的连接效率★：★异步明显更好★                 │
  │   （同步 + 线程池受 40 线程限制，异步能撑更多并发）    │
  │ ★CPU 开销★：★异步略高★（greenlet 切换 + 事件循环）    │
  │ ★开发复杂度★：★异步明显更高★                        │
  └────────────────────────────────────────────────────┘
  ★ ★判断：★
    ✓ ★高并发 + 短查询 + IO 密集 → 异步值得★
    ✗ ★查询本身很慢（几百毫秒）→ 异步帮不上忙★
    ✗ ★团队不熟悉 → 同步 + def 路由更省心★
  ★ ★同步 SQLAlchemy + def 路由在 FastAPI 里是完全正当的选择★

★ ★常见错误速查★：
  ┌────────────────────────────────────┬────────────────────┐
  │ MissingGreenlet                     │ ★懒加载/过期属性★   │
  │ ...concurrent operations not        │ ★共用 session 并发★ │
  │ permitted                           │                     │
  │ The unique() method must be invoked │ ★joinedload 一对多★ │
  │ Instance is not bound to a Session  │ ★session 已关闭★    │
  │ 对象在 commit 后属性全没了           │ ★expire_on_commit★  │
  │ Event loop is closed（退出时）       │ ★没 engine.dispose★ │
  │ too many connections                │ ★workers×pool 超了★ │
  │ prepared statement does not exist   │ ★PgBouncer + 缓存★  │
  └────────────────────────────────────┴────────────────────┘

★ ★N+1 的排查★：
  create_async_engine(url, ★echo=True★)     # ★开发时看 SQL★
  # 或统计条数
  from sqlalchemy import event
  @event.listens_for(engine.sync_engine, "before_cursor_execute")
  def count_sql(conn, cursor, stmt, params, context, executemany):
      ★request_sql_count[get_request_id()] += 1★
  ★ ★中间件里超过阈值就告警★

★ ★批量操作★：
  # ★批量插入★
  await session.execute(insert(User), [{"name": "a"}, {"name": "b"}])
  # ★或 add_all（★会构造 ORM 对象，慢一些但有事件★）★
  session.add_all([User(name="a"), User(name="b")])
  # ★批量更新★
  await session.execute(
      update(User), [{"id": 1, "name": "x"}, {"id": 2, "name": "y"}])
  ★ ★注意：bulk 操作不触发 ORM 事件、不更新 identity map★

★ ★只读查询的优化★：
  # ① 不构造 ORM 对象
  rows = (await session.execute(select(User.id, User.name))).all()
  # ② 或用 .mappings() 拿字典
  rows = (await session.execute(stmt)).mappings().all()
  # ③ 大结果集用流式
  result = await session.stream(select(User))
  async for row in result.scalars():
      process(row)                     # ★★不把全部读进内存★★

★ ★常被忽略的配置★：
  create_async_engine(url,
      pool_pre_ping=True,        # ★★检测断开的连接（MySQL 8 小时超时）★★
      pool_recycle=1800,         # ★主动回收，要小于数据库的 wait_timeout★
      connect_args={"server_settings": {"jit": "off"}},  # ★asyncpg + PG：
                                 #   关掉 JIT 有时能显著提速短查询★
      echo_pool="debug",         # 排查连接池问题
  )

关于「异步 ORM 是否更快」要诚实：单条查询的延迟几乎没区别（瓶颈在数据库），高并发下的连接效率异步明显更好（同步加线程池受 40 线程限制），但CPU 开销略高、开发复杂度明显更高——所以「同步 SQLAlchemy + def 路由」在 FastAPI 里是完全正当的选择，尤其当查询本身就要几百毫秒时，异步帮不上忙。错误速查表覆盖了八类典型问题。N+1 排查用 echo=True 或注册 before_cursor_execute 事件统计条数、中间件里超阈值告警。性能上还有几个常被忽略的配置：pool_pre_ping=True（治 MySQL 8 小时超时）、pool_recycle、以及 asyncpg + PG 时关掉 JIT 有时能显著提速短查询。

六、实践清单

★ 起步配置（照抄即可）：
  engine = create_async_engine(
      settings.async_database_url,
      pool_size=5, max_overflow=10,        # ★★算过 workers × 这个数★★
      pool_pre_ping=True, pool_recycle=1800,
      echo=settings.DEBUG,
  )
  AsyncSessionLocal = async_sessionmaker(
      engine, ★expire_on_commit=False★, autoflush=False)

  class Base(★AsyncAttrs★, DeclarativeBase): pass

  class User(Base):
      orders: Mapped[list["Order"]] = relationship(★lazy="raise"★)

★ 检查清单：
  □ ★expire_on_commit=False★
  □ ★所有 relationship 设 lazy="raise"（开发期暴露问题）★
  □ ★查询时显式 selectinload/joinedload★
  □ ★joinedload 一对多记得 .unique()★
  □ ★一个请求一个 session，不并发共用★
  □ ★事务边界在 service（async with session.begin()）★
  □ ★依赖里不 commit★
  □ ★lifespan 里 engine.dispose()★
  □ ★连接数算过：workers × (pool+overflow) < max_connections★
  □ ★PgBouncer 时 statement_cache_size=0★
  □ ★response_model 的字段都预加载了★
  □ ★迁移用 Alembic（可用同步驱动）★
  □ ★开发环境 echo=True 看 SQL 条数★

★ ★选型决策★：
  ┌──────────────────────────────┬────────────────────┐
  │ 高并发 + 短查询 + 全异步栈     │ ★异步 SQLAlchemy★   │
  │ 查询慢 / 团队不熟 / 遗留代码   │ ★同步 + def 路由★   │
  │ 需要 pandas/科学计算等同步库   │ ★同步★              │
  │ 只读为主、简单查询             │ ★两者都行★          │
  └──────────────────────────────┴────────────────────┘
  ★ ★别因为"FastAPI 是异步框架"就强上异步 ORM★

★ 一句话总结：
  ★"SQLAlchemy 的异步是用 greenlet 把同步 ORM 内核桥接到 asyncio，
    所以任何『不经 await 的隐式 IO』都会抛 MissingGreenlet——
    最典型的是懒加载和 commit 后的过期属性；
    对策是 expire_on_commit=False、relationship 设 lazy='raise'、
    查询时显式预加载；session 不能并发共用，事务边界放 service。"★

起步配置可以照抄。检查清单里最关键的四条：expire_on_commit=False、所有 relationship 设 lazy="raise"、查询时显式预加载、连接数算过。选型上要清醒：别因为「FastAPI 是异步框架」就强上异步 ORM——查询慢、团队不熟、有遗留同步代码时，同步 SQLAlchemy 配 def 路由是完全正当的选择。

记忆钩子：「★SQLAlchemy 2.0 的异步不是把 ORM 重写成了 async，而是用 greenlet 把十几年积累的同步 ORM 内核桥接到 asyncio★——await session.execute() 会 greenlet_spawn 进入一个 greenlet，同步内核在里面工作，★需要真正发 SQL 时把控制权抛回事件循环去 await asyncpg★。★理解这点就理解了 MissingGreenlet：只要在『不经 await 的地方』触发了数据库 IO 就会抛它★——★最典型的是懒加载（user.orders 是纯同步的属性访问，没有 greenlet 上下文）★和★commit 后访问被 expire 的属性★，还有★Pydantic 序列化时遍历字段★、★日志里 f’{obj}’ 触发 repr★、列表推导里访问关系。★三条铁律★：★① expire_on_commit=False 几乎必配★（默认 commit 后所有实例标记过期，下次访问属性要重新查库——同步下只是多一次查询，★异步下就是不经 await 的 IO 直接炸★，典型场景是 service commit 后 return order 交给 response_model 序列化）；★② 关系必须预加载，并给所有 relationship 设 lazy=‘raise’★（把运行时的 MissingGreenlet 变成开发期就能发现的明确报错）；★③ 只能用 2.0 风格的 select() + await session.execute()★，旧的 Query API 在异步下不可用，★取对象要 .scalars()★。预加载选型：★selectinload 用于一对多/多对多（额外一条 IN 查询、不产生笛卡尔积、可嵌套）、joinedload 用于多对一/一对一（一条 SQL，★但一对多时行数放大且必须 .unique() 否则报错★）、contains_eager 用于要在 JOIN 上加条件时★。★AsyncSession 不是并发安全的★（内部有 identity map、待 flush 队列、事务状态），★asyncio.gather 里共用一个 session 会报 concurrent operations not permitted★——要么串行、要么各开 session（代价是不在同一事务）。★事务边界放 service 用 async with session.begin()★，★依赖里不要 commit★；★flush 发 SQL 但不提交、能拿自增 id★。工程细节：★lifespan 里要 await engine.dispose()★（否则退出时 Event loop is closed）、★生产别在 lifespan 里 create_all()，用 Alembic（迁移可以用同步驱动，运行时用异步驱动最省事）★、★run_sync 是调用同步 API 的桥梁★、★连接数 = workers × (pool + overflow) 容易超过 PG 默认的 100★、★PgBouncer transaction 模式要设 statement_cache_size=0★、★asyncpg 不支持 ?sslmode= 这类 libpq 参数（要走 connect_args）★。★最后要诚实：单条查询延迟异步和同步几乎没区别（瓶颈在数据库），异步的优势在高并发下的连接效率；查询本身慢、团队不熟、有同步遗留代码时，『同步 SQLAlchemy + def 路由走线程池』在 FastAPI 里是完全正当的选择★。」

七、常见误区与追问

• 误区：AsyncSession 就是把 Session 的方法都加上了 await，用法基本一样。 底层机制完全不同，带来的约束也不同。SQLAlchemy 并没有把 ORM 重写成异步——identity map、工作单元、关系加载器这些十几年积累的核心逻辑仍然是同步代码，异步支持是用 greenlet 做的桥接层：await session.execute() 时先 greenlet_spawn 创建一个 greenlet，同步内核在里面运行，当它需要真正发送 SQL 时，通过 await_only() 把控制权「抛」回事件循环去 await asyncpg 的真异步 IO，完成后再恢复 greenlet 继续处理结果。这个机制的硬性要求是「所有数据库 IO 必须发生在 greenlet 上下文里」——而懒加载（user.orders）是在纯粹的属性访问中发起 IO 的，此时根本没有 greenlet，于是抛出 MissingGreenlet: greenlet_spawn has not been called。所以异步下你必须显式管理「什么时候加载什么数据」，这是同步版本不需要操心的。
• 误区：MissingGreenlet 报错时，在报错的那一行加个 await 就能修好。 加不了 await，因为报错的位置通常是属性访问而不是函数调用（user.orders 你没法写成 await user.orders）。而且报错的位置往往不是问题的根源——真正的原因在更早的查询语句没有预加载关系。典型场景是：service 里 await db.get(User, uid) 拿到用户对象，路由把它返回给 response_model=UserOut，Pydantic 在序列化时遍历字段访问了 user.orders，报错的 traceback 指向的是 Pydantic 内部的代码。正确的修法有三种：① 在查询时预加载（select(User).options(selectinload(User.orders))，这是最正确的）；② 让输出模型不包含那个关系字段（如果本来就不需要）；③ 用 AsyncAttrs 显式加载（await user.awaitable_attrs.orders，但循环里用就是 N+1）。定位根源的最好办法是给所有 relationship 设 lazy="raise"——报错会立刻变成清晰的「哪个属性、哪个对象需要预加载」。
• 误区：expire_on_commit 是个小优化，用默认值就行。 在异步下它几乎是必配项。SQLAlchemy 的默认行为是 commit() 后把所有实例标记为「过期」——设计意图是「事务提交后，内存中的对象可能已经和数据库不一致，下次访问属性时重新查一遍最新值」。在同步代码里这只是多一次查询（无感）；在异步代码里，这次「重新查」是从属性访问触发的、不经过 await，于是直接抛 MissingGreenlet。最典型的翻车路径：service 里 session.add(order) → await session.commit() → return order；路由拿到这个 order 交给 response_model 序列化，访问 order.id 的瞬间就炸了——而且报错信息完全指不到 commit 那行。所以 async_sessionmaker(engine, expire_on_commit=False) 应该是默认配置。代价是：commit 后对象里的值是内存中的旧值，如果数据库有触发器或默认值改动了数据，需要显式 await session.refresh(obj)。
• 误区：用 asyncio.gather 并发执行多个查询能加速。 共用同一个 AsyncSession 会直接报错：InvalidRequestError: This session is provisioning a new connection; concurrent operations are not permitted。原因是 Session 不是并发安全的——它内部维护着 identity map（对象缓存）、待 flush 的变更队列、当前事务和连接的绑定关系，两个协程交错操作会破坏这些状态（更可怕的是某些情况下不报错但数据错乱）。如果确实需要并发查询，有两个选择：① 各自开独立的 session（async with AsyncSessionLocal() as s: 在每个协程里），代价是它们不在同一个事务里、并且会占用多个连接；② 老老实实串行——大多数场景下几条查询串行执行的总耗时也就几毫秒，并发带来的复杂度不值得。真正需要并发的是「同时调用多个外部服务」这类场景，而不是数据库查询。
• 误区：既然 FastAPI 是异步框架，数据库当然要用异步 SQLAlchemy。 这个推论并不成立，很多项目用同步 SQLAlchemy 反而更合适。诚实的对比是：单条查询的延迟，异步和同步几乎没有区别（瓶颈是数据库执行 SQL 的时间，不是驱动）；异步真正的优势是高并发下的连接效率——同步方案下每个请求要占一个线程，而 FastAPI 的默认线程池只有 40 个，异步则能轻松撑起几百上千并发。但异步的代价也很实在：必须处理预加载、MissingGreenlet、session 不能并发共用、Alembic 配置更麻烦、可用的库更少（很多分析工具、报表库、第三方 SDK 都是同步的）。判断标准：高并发 + 短查询 + 全异步技术栈 → 异步值得；查询本身就要几百毫秒（复杂报表）、团队不熟悉、有大量同步遗留代码 → 同步 SQLAlchemy 配 def 路由（走线程池）是完全正当的选择，FastAPI 官方文档也明确支持这种用法。
• 追问：selectinload 和 joinedload 具体怎么选？ 按关系的方向选。joinedload 生成一条带 LEFT OUTER JOIN 的 SQL——适合多对一/一对一（Order.user、User.profile），因为每个订单只有一个用户，JOIN 不会让结果集变大。但用在一对多上会产生笛卡尔积放大：加载 100 个用户及其各 10 个订单，JOIN 返回 1000 行（用户的字段重复 10 次），网络传输和 Python 侧的去重都变贵；而且必须调用 .unique()，否则 SQLAlchemy 会直接报 InvalidRequestError: The unique() method must be invoked on this Result（因为它无法自动判断是否需要去重）。selectinload 则发两条 SQL：先查 100 个用户，再用 WHERE user_id IN (...) 一次性把所有订单查出来，行数不放大，而且嵌套时更清晰（selectinload(User.orders).selectinload(Order.items)）。经验法则：一对多、多对多用 selectinload，多对一、一对一用 joinedload，两者可以在同一个查询里组合。第三个选项 contains_eager 用于「需要在 JOIN 上加过滤条件」的场景（只加载已支付的订单），但要注意此时 user.orders 里就只有部分数据了。
• 追问：lazy="raise" 值得全项目开启吗？ 值得，而且是异步 SQLAlchemy 项目最有价值的一条配置。它的作用是：任何未经预加载就访问关系属性的行为，都会立刻抛出 InvalidRequestError: 'User.orders' is not available due to lazy='raise'——报错清晰地指出了是哪个模型的哪个属性，比 MissingGreenlet 那句含糊的「IO attempted in an unexpected place」好定位一百倍。更重要的是它把问题提前到了开发期：本地跑一遍就能发现所有遗漏预加载的地方，而不是等到线上某个特定路径被触发才炸。它还有一个附带好处——在同步项目里同样能防 N+1（懒加载不会静默地发出几百条 SQL，而是直接报错逼你显式预加载）。如果觉得全局 raise 太严格，可以用 lazy="raise_on_sql"：已经在 identity map 里的数据仍可访问，只有真正需要发 SQL 时才报错。查询级也有对应的 raiseload("*")，可以在特定查询里禁掉所有未声明的加载。
• 追问：Alembic 在异步项目里怎么配？ 有两条路，推荐第二条。① 完全异步的 env.py——需要改写 run_migrations_online()：用 create_async_engine 建立连接，然后 await conn.run_sync(do_run_migrations) 把同步的 Alembic API 桥接进来，最后 asyncio.run(...) 驱动整个流程；Alembic 的模板里有现成的 async 版本（alembic init -t async）。这条路的好处是应用和迁移用同一套配置，坏处是 env.py 变复杂，而且某些 Alembic 的自动生成功能在异步下有额外的坑。② 迁移用同步驱动、运行时用异步驱动——alembic.ini 里配 postgresql://...（走 psycopg2），应用里配 postgresql+asyncpg://...。这是更简单也更常见的做法：迁移本来就是一次性的离线操作，完全没有异步的必要；代价只是多装一个同步驱动。实践中还有个小技巧：在 Settings 里放一个 database_url，然后用 computed_field 派生出 async_database_url（把 scheme 替换掉），这样两边配置不会漂移。

八、加强记忆

SQLAlchemy 2.0 的异步不是把 ORM 重写成了 async，而是用 greenlet 把十几年积累的同步 ORM 内核桥接到 asyncio——await session.execute() 会 greenlet_spawn 进入一个 greenlet，同步内核在里面工作，需要真正发 SQL 时把控制权抛回事件循环去 await asyncpg。理解这一点就理解了 MissingGreenlet：只要在「不经 await 的地方」触发了数据库 IO 就会抛它——最典型的是懒加载（user.orders 是纯同步的属性访问，没有 greenlet 上下文）和 commit 后访问被 expire 的属性，此外还有 Pydantic 序列化时遍历字段、日志里 f"{obj}" 触发 __repr__、列表推导里访问关系。三条铁律：① expire_on_commit=False 几乎必配（默认 commit 后所有实例标记过期，下次访问属性要重新查库——同步下只是多一次查询，异步下就是不经 await 的 IO 直接炸，典型场景是 service commit 后 return order 交给 response_model 序列化）；② 关系必须预加载，并给所有 relationship 设 lazy="raise"（把运行时的 MissingGreenlet 变成开发期就能发现的明确报错）；③ 只能用 2.0 风格的 select() + await session.execute()，旧的 Query API 在异步下不可用，取对象要 .scalars()。预加载选型：selectinload 用于一对多/多对多（额外一条 IN 查询、不产生笛卡尔积、可嵌套）、joinedload 用于多对一/一对一（一条 SQL，但一对多时行数放大且必须 .unique() 否则报错）、contains_eager 用于需要在 JOIN 上加条件时。AsyncSession 不是并发安全的（内部有 identity map、待 flush 队列、事务状态），asyncio.gather 里共用一个 session 会报 concurrent operations are not permitted——要么串行、要么各开一个 session（代价是不在同一事务里）。事务边界放在 service 层用 async with session.begin()，依赖里不要 commit；flush 发 SQL 但不提交、能拿到自增 id。工程细节：lifespan 里要 await engine.dispose()（否则退出时报 Event loop is closed）、生产别在 lifespan 里 create_all()，用 Alembic（迁移用同步驱动、运行时用异步驱动最省事）、run_sync 是调用同步 API 的桥梁、连接数 = workers × (pool + overflow) 很容易超过 PG 默认的 100、PgBouncer 的 transaction 模式要设 statement_cache_size=0、asyncpg 不支持 ?sslmode= 这类 libpq 参数（要走 connect_args）。最后要诚实：单条查询的延迟异步和同步几乎没区别（瓶颈在数据库），异步的优势在高并发下的连接效率；查询本身就慢、团队不熟悉、有同步遗留代码时，「同步 SQLAlchemy + def 路由走线程池」在 FastAPI 里是完全正当的选择。