我们从Python开源项目中，提取了以下15个代码示例，用于说明如何使用django.db.transaction.set_rollback()。
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 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 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 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 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})
