Question about nested transaction (savepoint)
Closed this issue · 2 comments
op13op commented
Hello, I'm trying to use nested transaction in pytest fixture to achieve auto rollback in test environment. And I notice asyncpg support savepiont, but I just could not make it.
The following is code snippet:
class ExplilcitRollbackException(Exception):
pass
@pytest.fixture
async def pg_pool(loop, pg_setting, setup_databse):
from asyncpgsa import create_pool
pool = await create_pool(**pg_setting, loop=loop)
try:
async with pool.transaction():
yield pool
raise ExplilcitRollbackException
except ExplilcitRollbackException:
pass
finally:
await pool.close()Are there any ways to achieve this, or I could only truncate the table?
allburov commented
@op13op have you resolved issue?
If I am not wrong, pool.transaction return a context manager with a pair - connection + transaction. In above example you have two different connections with two transaction. It's a reason why DB wasn't rollbacked after first test.
You should create single connection and create two transaction:
pool = await create_pool(**pg_setting, loop=loop)
async with pool.acquire() as con:
async with con.transaction():
async with con.transaction():
yield conn # You MUST use this connection instance in all your code. May be monkeypatch you module or directly pass them. Otherwise you will create new connection without transaction.