主页 / 问题 / 1552127
Accepted
slx0009
Asked:2023-11-20 01:26:26 +0800 CST2023-11-20 01:26:26 +0800 CST

如何返回异步 SQLAlchemy 会话？

  • 772

我有一个带有异步会话生成器的 Python 数据库类。

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 get_session_maker(self) -> sessionmaker:
        return self._session_maker

我有一个从该数据库接收数据的存储库。

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_maker()() as session:
            return await session.get(User, entity_id)

一切工作正常，但我不喜欢该方法的工作原理和外观get_session_maker。可以做这样的事情吗？

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)

    async def get_session(self) -> AsyncSession:
        async with self._session_maker() as session:
            yield session


class UsersRepository(AbstractUsersRepository):
    def __init__(self, db: AbstractDatabase):
        self._db = db

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

但我无法执行这个确切的选项，因为我收到错误TypeError: object async_generator can't be used in 'await' expression。我真的不明白为什么。它不是这样工作的：вызывается get-> get_session начинает сессию и возвращает ее-> get закончил выполнение->вернулся к get_session и закрыл сессию吗？

我也尝试过这个选项，但它也不起作用。

async def get_session(self) -> AsyncSession:
    session: AsyncSession = self._session_maker()
    try:
        yield session
    except Exception as e:
        await session.rollback()
        print(f"An error occurred: {e}")
    finally:
        await session.close()
python

1 个回答

  1. Best Answer

    到目前为止，我能够实现的最佳解决方案是工作单元模式。

    class UnitOfWork(AbstractUnitOfWork):
    def __init__(self) -> None:
        self._session_maker = SESSION_MAKER

    async def __aenter__(self) -> AbstractUnitOfWork:
        self._session = self._session_maker()
        self.repository1 = Repository1(self._session)
        self.repository2 = Repository2(self._session)
        return self

    async def __aexit__(self, *args) -> None:
        await self._session.rollback()
        await self._session.close()

    async def commit(self) -> None:
        await self._session.commit()

    这是带有数据库的文件。

    ENGINE = AsyncEngine(create_engine("postgresql+asyncpg://postgres:postgres@localhost/test", echo=True, future=True))
SESSION_MAKER = sessionmaker(ENGINE, class_=AsyncSession, expire_on_commit=False)

    这种实现的优点是我可以使用一个会话访问多个存储库，这将允许我在发生错误时回滚必须发送到数据库的任何更改。

    在我看来，缺点是__aenter__存储库不断被重新创建。因此，最好在构造函数中进行它们的初始化，而不传输会话。当它被调用时，__aenter__您可以创建一个会话并将其传输到存储库。好吧，__aexit__关闭后，清理一下存储库。

    UnitOfWork您可以在服务层上使用它。

    class StackOverflowService(AbstractStackOverflowService):
    def __init__(self, uow: AbstractUnitOfWork):
        self._uow = uow

    async def do_something(self) -> None:
        async with self._uow as uow:
            await uow.repository1.delete_moderator('Михаил Ребров')
            await uow.repository1.add_user('Михаил Ребров')
            await uow.commit()
    • 0

相关问题