Method core_session

api/oss/src/dbs/postgres/shared/engine.py:71–83

(self)

    @asynccontextmanager
    async def core_session(self) -> AsyncGenerator[AsyncSession, None]:
        session: AsyncSession = self.async_core_session()

        try:
            yield session
            await session.commit()

        except Exception as e:
            await session.rollback()
            raise e

        finally:
            await session.close()

    @asynccontextmanager
    async def tracing_session(self) -> AsyncGenerator[AsyncSession, None]: