INNER CODE UNIT · Python

async_engine

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

    def async_engine(async_sqlalchemy_url: str) -> AsyncEngine:
        return create_async_engine(async_sqlalchemy_url)

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

View source record →