INNER CODE UNIT · Python
patch_new_async_engine
dialoguemd/fastapi-sqla · fastapi_sqla/_pytest_plugin.py:172
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,
):
yield
await transaction.rollback()
@fixture
async def async_sqla_reflection(