Eg, I want this function:
from django.db import transaction
with transaction.atomic():
assert inside_transaction() == True
assert inside_transaction() == False
Is there a way to achieve this? Or is it possible to detect directly in psycopg2 if not in Django's ORM?
CodePudding user response:
Looking at the source code [GitHub] for the Atomic
class that is used by transaction.atomic
it sets a flag in_atomic_block
on the connection. Hence if you want to check if your code is inside a transaction you can write the follows:
with transaction.atomic():
assert transaction.get_connection().in_atomic_block == True
assert transaction.get_connection().in_atomic_block == False
Note: The get_connection
function optionally takes a parameter using
, in case you specify using
for the call to transaction.atomic
, you should pass it to get_connection
as well.
Better yet since get_connection
is a private API just write a small function of your own to check this:
from django.db import transaction
from django.db import (
DEFAULT_DB_ALIAS, connections,
)
def inside_transaction(using=None):
if using is None:
using = DEFAULT_DB_ALIAS
return connections[using].in_atomic_block
with transaction.atomic():
assert inside_transaction() == True
assert inside_transaction() == False