INNER CODE UNIT · Python

async_session_factory

dialoguemd/fastapi-sqla · fastapi_sqla/_pytest_plugin.py:197

    def async_session_factory(
        async_sqla_connection: AsyncConnection,
        async_sqla_reflection,
        patch_new_async_engine,
    ) -> sessionmaker:
        # TODO: Use async_sessionmaker once only supporting 2.x+
        return sessionmaker(
            bind=async_sqla_connection, expire_on_commit=False, class_=AsyncSession
        )  # type: ignore

    @fixture
    async def async_session(
        async_session_factory: sessionmaker,
    ) -> AsyncGenerator[AsyncSession]:
        session: AsyncSession = async_session_factory()
        yield session
        await session.close()

View source record →