如何使用AsyncSession.sync_session？同步异步会话同事务实现

问题根源 #

1. AsyncSession.sync_session 仅能在异步事件循环的greenlet上下文中访问，直接在同步代码（如Factory Boy工厂方法）中调用会触发MissingGreenlet错误。
2. 要实现同步/异步会话共享同一事务，必须让两者绑定同一个数据库连接与事务上下文。

解决方案一：手动绑定同一连接实现事务共享 #

通过异步连接的sync_connection属性获取同步连接，分别创建异步/同步会话，确保两者共用同一事务：

1. 数据库配置 #

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import Session, sessionmaker

# 异步引擎
ASYNC_DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
async_engine = create_async_engine(ASYNC_DATABASE_URL, echo=True)

# 异步会话工厂
AsyncSessionLocal = sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False
)

2. 共享事务的操作示例 #

from factories import UserFactory  # 你的Factory Boy工厂

async def create_user_and_query():
    # 开启异步连接与事务
    async with async_engine.connect() as async_conn:
        async with async_conn.begin():
            # 获取同步连接，绑定到同一物理连接
            sync_conn = async_conn.sync_connection
            
            # 创建同步会话，绑定到同步连接
            sync_session = Session(bind=sync_conn, expire_on_commit=False)
            # 创建异步会话，绑定到异步连接
            async_session = AsyncSession(bind=async_conn, expire_on_commit=False)
            
            # 用Factory Boy（同步）创建对象
            user = UserFactory(session=sync_session)
            sync_session.flush()  # 写入事务，不提交
            
            # 用异步会话查询同一对象
            db_user = await async_session.get(User, user.id)
            print(f"异步查询结果：{db_user.name}")
            
            # 提交事务（异步连接的提交会同步到同步会话）
            await async_conn.commit()
            
            # 关闭会话
            sync_session.close()
            await async_session.close()

3. FastAPI路由集成 #

在FastAPI中通过依赖注入实现共享事务：

from fastapi import FastAPI, Depends

app = FastAPI()

async def get_shared_sessions():
    async with async_engine.connect() as async_conn:
        async with async_conn.begin():
            sync_conn = async_conn.sync_connection
            sync_session = Session(bind=sync_conn, expire_on_commit=False)
            async_session = AsyncSession(bind=async_conn, expire_on_commit=False)
            yield async_session, sync_session
            sync_session.close()
            await async_session.close()

@app.post("/users/")
async def create_user(
    sessions: tuple[AsyncSession, Session] = Depends(get_shared_sessions)
):
    async_session, sync_session = sessions
    # 用Factory Boy创建对象
    user = UserFactory(session=sync_session)
    sync_session.flush()
    # 异步会话操作对象
    await async_session.refresh(user)
    return {"id": user.id, "name": user.name}

解决方案二：在greenlet上下文运行同步代码 #

如果不需要手动管理连接，可直接在异步上下文内获取AsyncSession.sync_session，并确保同步代码运行在greenlet上下文中：

1. 修复Factory Boy调用方式 #

在异步函数（如FastAPI路由、异步测试用例）中直接调用Factory Boy：

@app.post("/users/")
async def create_user(session: AsyncSession = Depends(get_async_session)):
    # 获取同步会话（当前在异步greenlet上下文内，不会触发MissingGreenlet）
    sync_session = session.sync_session
    
    # 直接运行同步的Factory Boy代码
    user = UserFactory(session=sync_session)
    await session.commit()
    await session.refresh(user)
    return {"id": user.id, "name": user.name}

注意事项 #

• 此方式下同步代码会阻塞事件循环，若Factory Boy操作耗时较长，建议用asyncio.to_thread（但需注意事务共享问题，to_thread会切换线程，无法直接复用当前异步会话的连接）。
• 确保运行环境支持greenlet（FastAPI/Starlette默认使用anyio，已支持）。

关键注意点 #

• 无论哪种方案，不要在纯同步代码中直接访问AsyncSession.sync_session，必须在异步上下文内操作。
• 共享事务时，所有操作（同步/异步）都需在同一连接的事务上下文中执行，避免出现数据不一致。
• Factory Boy的工厂需配置使用传入的会话，而非自行创建新会话：

  # Factory Boy示例配置
  import factory
  from models import User
  
  class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
      class Meta:
          model = User
          sqlalchemy_session = None  # 不固定会话，使用传入的session
          sqlalchemy_session_persistence = "flush"  # 仅flush，不自动提交
  
      name = factory.Faker("name")
      email = factory.Faker("email")
  

内容的提问来源于stack exchange，提问作者Альберт Александров