如何正确返回SQLAlchemy异步会话？现有实现优化疑问

报错原因 #

你编写的get_session方法因使用yield关键字，成为了异步生成器。异步生成器属于可迭代对象，不能通过await直接获取返回值，必须使用async for循环遍历或调用anext()方法，直接await会触发TypeError: object async_generator can't be used in 'await' expression。

优化方案 #

方案一：异步上下文管理器（推荐） #

使用contextlib.asynccontextmanager装饰器将get_session改造为异步上下文管理器，既简化调用，又能自动管理会话的生命周期：

from contextlib import asynccontextmanager

class Database(AbstractDatabase):
    def __init__(self):
        self._engine = AsyncEngine(create_engine("postgresql+asyncpg://postgres:postgres@localhost/test", echo=True, future=True))
        self._session_maker = sessionmaker(self._engine, class_=AsyncSession, expire_on_commit=False)

    @asynccontextmanager
    async def get_session(self):
        async with self._session_maker() as session:
            yield session

# UsersRepository 调用示例
class UsersRepository(AbstractUsersRepository):
    def __init__(self, db: AbstractDatabase):
        self._db = db

    async def get(self, entity_id: UUID) -> User:
        async with self._db.get_session() as session:
            return await session.get(User, entity_id)

方案二：简化会话生成器调用 #

给Database类添加一个快捷方法，直接返回会话实例的上下文，减少冗余的括号调用：

class Database(AbstractDatabase):
    def __init__(self):
        self._engine = AsyncEngine(create_engine("postgresql+asyncpg://postgres:postgres@localhost/test", echo=True, future=True))
        self._session_maker = sessionmaker(self._engine, class_=AsyncSession, expire_on_commit=False)

    def session(self):
        return self._session_maker()

# UsersRepository 调用示例
class UsersRepository(AbstractUsersRepository):
    def __init__(self, db: AbstractDatabase):
        self._db = db

    async def get(self, entity_id: UUID) -> User:
        async with self._db.session() as session:
            return await session.get(User, entity_id)

方案三：让Database类本身成为异步上下文管理器 #

适合短生命周期的数据库实例场景，直接通过async with获取会话：

class Database(AbstractDatabase):
    def __init__(self):
        self._engine = AsyncEngine(create_engine("postgresql+asyncpg://postgres:postgres@localhost/test", echo=True, future=True))
        self._session_maker = sessionmaker(self._engine, class_=AsyncSession, expire_on_commit=False)
        self._session = None

    async def __aenter__(self):
        self._session = self._session_maker()
        await self._session.__aenter__()
        return self._session

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self._session.__aexit__(exc_type, exc_val, exc_tb)

# 调用示例
async def fetch_user(user_id: UUID):
    async with Database() as session:
        return await session.get(User, user_id)

内容的提问来源于stack exchange，提问作者sln