INNER CODE UNIT · Python

async_sqla_reflection

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

    async def async_sqla_reflection(
        sqla_modules, async_sqla_connection: AsyncConnection
    ):
        from fastapi_sqla import Base  # noqa: PLC0415

        await async_sqla_connection.run_sync(lambda conn: Base.prepare(conn.engine))

    @fixture
    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

View source record →