test_transactions.py 文件源码

def test_atomic_with_delete(flushdb):
    for i in range(3):
        await User.create(username=f'u{i}')

    async with db.atomic():
        user = await User.get(User.username == 'u1')
        await user.delete_instance()

    usernames = [u.username async for u in User.select()]
    assert sorted(usernames) == ['u0', 'u2']

    async with db.atomic():
        async with db.atomic():
            user = await User.get(User.username == 'u2')
            await user.delete_instance()

    usernames = [u.username async for u in User.select()]
    assert usernames == ['u0']