INNER CODE UNIT · Python

async_sqla_connection

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

    async def async_sqla_connection(
        async_engine: AsyncEngine,
    ) -> AsyncGenerator[AsyncConnection]:
        async with async_engine.connect() as connection:
            yield connection

    @fixture
    async def patch_new_async_engine(
        request: FixtureRequest, async_sqla_connection: AsyncConnection
    ):
        """So that all async DB operations are never written to db for real."""
        if "dont_patch_engines" in request.keywords:
            yield
        else:
            async with async_sqla_connection.begin() as transaction:
                with patch(
                    "fastapi_sqla.async_sqla.new_async_engine",
                    return_value=async_sqla_connection,

View source record →