def _rollback_atomics(cls, atomics):
"""Rollback atomic blocks opened through the previous method"""
for db_name in reversed(cls._databases_names()):
transaction.set_rollback(True, using=db_name)
atomics[db_name].__exit__(None, None, None)
python类set_rollback()的实例源码
def set_rollback():
if hasattr(transaction, 'set_rollback'):
if connection.settings_dict.get('ATOMIC_REQUESTS', False):
# If running in >=1.6 then mark a rollback as required,
# and allow it to be handled by Django.
if connection.in_atomic_block:
transaction.set_rollback(True)
elif transaction.is_managed():
# Otherwise handle it explicitly if in managed mode.
if transaction.is_dirty():
transaction.rollback()
transaction.leave_transaction_management()
else:
# transaction not managed
pass
def set_rollback():
if hasattr(transaction, 'set_rollback'):
if connection.settings_dict.get('ATOMIC_REQUESTS', False):
# If running in >=1.6 then mark a rollback as required,
# and allow it to be handled by Django.
if connection.in_atomic_block:
transaction.set_rollback(True)
elif transaction.is_managed():
# Otherwise handle it explicitly if in managed mode.
if transaction.is_dirty():
transaction.rollback()
transaction.leave_transaction_management()
else:
# transaction not managed
pass
def set_rollback():
if hasattr(transaction, 'set_rollback'):
if connection.settings_dict.get('ATOMIC_REQUESTS', False):
# If running in >=1.6 then mark a rollback as required,
# and allow it to be handled by Django.
if connection.in_atomic_block:
transaction.set_rollback(True)
elif transaction.is_managed():
# Otherwise handle it explicitly if in managed mode.
if transaction.is_dirty():
transaction.rollback()
transaction.leave_transaction_management()
else:
# transaction not managed
pass
def test_first_access_with_rollback(self):
def one(output):
with transaction.atomic():
output.append(('one', 'begin'))
value = get_next_value()
output.append(('one', value))
time.sleep(0.2)
transaction.set_rollback(True)
output.append(('one', 'rollback'))
connection.close()
def two(output):
time.sleep(0.1)
with transaction.atomic():
output.append(('two', 'begin'))
value = get_next_value()
output.append(('two', value))
output.append(('two', 'commit'))
connection.close()
expected = [
('one', 'begin'),
('one', 1),
('two', 'begin'),
('one', 'rollback'),
('two', 1),
('two', 'commit'),
]
self.assertSequence(one, two, expected)
def test_later_access_with_rollback(self):
get_next_value()
def one(output):
with transaction.atomic():
output.append(('one', 'begin'))
value = get_next_value()
output.append(('one', value))
time.sleep(0.2)
transaction.set_rollback(True)
output.append(('one', 'rollback'))
connection.close()
def two(output):
time.sleep(0.1)
with transaction.atomic():
output.append(('two', 'begin'))
value = get_next_value()
output.append(('two', value))
output.append(('two', 'commit'))
connection.close()
expected = [
('one', 'begin'),
('one', 2),
('two', 'begin'),
('one', 'rollback'),
('two', 2),
('two', 'commit'),
]
self.assertSequence(one, two, expected)
def set_rollback():
if hasattr(transaction, 'set_rollback'):
if connection.settings_dict.get('ATOMIC_REQUESTS', False):
# If running in >=1.6 then mark a rollback as required,
# and allow it to be handled by Django.
if connection.in_atomic_block:
transaction.set_rollback(True)
elif transaction.is_managed():
# Otherwise handle it explicitly if in managed mode.
if transaction.is_dirty():
transaction.rollback()
transaction.leave_transaction_management()
else:
# transaction not managed
pass
def get_condition_result(condition, request=None):
template = Template(CONDITION_TEMPLATE.format(condition))
# Always assume that the end-user is dumb
with transaction.atomic():
try:
return template.render(context=RequestContext(request=request))
finally:
transaction.set_rollback(rollback=True)
def _rollback_atomics(cls, atomics):
"""Rollback atomic blocks opened through the previous method"""
for db_name in reversed(cls._databases_names()):
transaction.set_rollback(True, using=db_name)
atomics[db_name].__exit__(None, None, None)
def _rollback_atomics(cls, atomics):
"""Rollback atomic blocks opened through the previous method"""
for db_name in reversed(cls._databases_names()):
transaction.set_rollback(True, using=db_name)
atomics[db_name].__exit__(None, None, None)
def _rollback_atomics(cls, atomics):
"""Rollback atomic blocks opened through the previous method"""
for db_name in reversed(cls._databases_names()):
transaction.set_rollback(True, using=db_name)
atomics[db_name].__exit__(None, None, None)
def _rollback_atomics(cls, atomics):
"""Rollback atomic blocks opened through the previous method"""
for db_name in reversed(cls._databases_names()):
transaction.set_rollback(True, using=db_name)
atomics[db_name].__exit__(None, None, None)
def handle_exception(self, *args, **kwargs):
"""Handle exception with transaction rollback."""
response = super(AtomicMixin, self).handle_exception(*args, **kwargs)
if getattr(response, 'exception'):
# We've suppressed the exception but still need to rollback any transaction.
transaction.set_rollback(True)
return response
def _rollback_atomics(cls, atomics):
"""Rollback atomic blocks opened through the previous method"""
for db_name in reversed(cls._databases_names()):
transaction.set_rollback(True, using=db_name)
atomics[db_name].__exit__(None, None, None)
def run_test(self, request, object_id=None):
"""
Takes an HttpRequest with data from a ConfigToolForm and the
object id for a Model instance. Creates a temporary version
of the Model instance using the form data and returns a
JsonResponse with the result of the configuration test.
"""
try:
with transaction.atomic():
forms_are_valid = True
form = self._create_form_instance(request, object_id)
if form.is_valid():
instance = form.save()
# pass the parent model instance to the formsets to create
# related instances
formsets = self._create_formset_instances(request, instance)
if self._formsets_are_valid(formsets):
for formset in formsets:
formset.save()
# all models are now saved, so get the test result
result = self._get_result(form, instance)
else:
forms_are_valid = False
else:
formsets = []
forms_are_valid = False
if not forms_are_valid:
result = self._format_errors(form, formsets)
# rollback the database when exiting the atomic block
transaction.set_rollback(True)
except IntegrityError as error:
LOGGER.error('An error occurred while creating a test instance: %s', request)
result = 'Could not create an object for testing: %s' % error
except ValidationError as error:
LOGGER.error('An error occurred while initializing a config test: %s', request)
result = 'A validation error occurred: %s' % error
return JsonResponse({'result': result})
