Python django.db.transaction 模块，savepoint() 实例源码

def save(self, *args, **kwargs):
        sid = transaction.savepoint()
        if self.pk is None:
            i = 1
            while self.pk is None:

                # Protection from infinite loop
                if i > 20:
                    raise IntegrityError('Too many iterations while generating unique Invoice number.')

                self.payment_no = random.randint(1, 2147483646)

                try:
                    super(Invoice, self).save(*args, **kwargs)
                except IntegrityError:
                    transaction.savepoint_rollback(sid)

                i += 1
        else:
            super(Invoice, self).save(*args, **kwargs)

        transaction.savepoint_commit(sid)
        transaction.commit()

def forwards(self, orm):
        from sentry.db.models import create_or_update
        from sentry.utils.query import RangeQuerySetWrapperWithProgressBar

        OrganizationMember = orm['sentry.OrganizationMember']
        Team = orm['sentry.Team']

        queryset = Team.objects.all()

        for team in RangeQuerySetWrapperWithProgressBar(queryset):
            sid = transaction.savepoint()
            try:
                OrganizationMember.objects.create(
                    organization_id=team.organization_id,
                    user_id=team.owner_id,
                    type=0,  # OWNER
                )
            except IntegrityError:
                transaction.savepoint_rollback(sid)
            else:
                transaction.savepoint_commit(sid)
            transaction.commit()

def form_valid(self, form):
        # create a new user for this sponsor
        try:
            sp = transaction.savepoint()
            baseContact = form.cleaned_data["sponsorContact"]

            try:
                user = User.objects.get(username=form.cleaned_data["sponsorUsername"])
            except User.DoesNotExist:
                user = User(username = form.cleaned_data["sponsorUsername"])
                user.first_name = baseContact.contactPersonFirstname
                user.last_name = baseContact.contactPersonSurname
                user.email = baseContact.contactPersonEmail
                user.save()

                profile = UserProfile(user = user)
                profile.authToken = id_generator(24)
                profile.save()

            sponsoring = Sponsoring()
            sponsoring.owner = user
            sponsoring.year = getActiveYear(self.request)
            sponsoring.contact = baseContact
            sponsoring.package = form.cleaned_data["sponsorPackage"]
            sponsoring.adminComment = form.cleaned_data["internalComment"]
            sponsoring.save()

            self.object = sponsoring

            transaction.savepoint_commit(sp)
        except Exception as e:
            transaction.savepoint_rollback(sp)
            raise e


        return HttpResponseRedirect(self.get_success_url())

def before_scenario(context, _):
    context.atomic = transaction.Atomic(savepoint=True, using='default')
    context.atomic.__enter__()
    context._savepoint = transaction.savepoint()

def atomic(using=None):
        sid = transaction.savepoint(using=using)
        try:
            yield
        except IntegrityError:
            transaction.savepoint_rollback(sid, using=using)
            raise
        else:
            transaction.savepoint_commit(sid, using=using)

def forwards(self, orm):
        Organization = orm['sentry.Organization']
        OrganizationMember = orm['sentry.OrganizationMember']
        OrganizationMemberTeam = orm['sentry.OrganizationMemberTeam']
        Team = orm['sentry.Team']

        for org in Organization.objects.all():
            members = OrganizationMember.objects.filter(
                organization=org,
                has_global_access=True,
            )
            teams = Team.objects.filter(organization=org)
            for member in members:
                for team in teams:
                    # XXX(dcramer): South doesnt like us using transactions here
                    try:
                        sid = transaction.savepoint()
                        OrganizationMemberTeam.objects.create(
                            team=team,
                            organizationmember=member,
                            is_active=True,
                        )
                    except IntegrityError:
                        transaction.savepoint_rollback(sid)
                    else:
                        transaction.savepoint_commit(sid)
                    transaction.commit()

def save(self, actor, organization, ip_address):
        om = super(InviteOrganizationMemberForm, self).save(commit=False)
        om.organization = organization

        try:
            existing = OrganizationMember.objects.filter(
                organization=organization,
                user__email__iexact=om.email,
                user__is_active=True,
            )[0]
        except IndexError:
            pass
        else:
            return existing, False

        sid = transaction.savepoint(using='default')
        try:
            om.save()
        except IntegrityError:
            transaction.savepoint_rollback(sid, using='default')
            return OrganizationMember.objects.get(
                email__iexact=om.email,
                organization=organization,
            ), False
        transaction.savepoint_commit(sid, using='default')

        AuditLogEntry.objects.create(
            organization=organization,
            actor=actor,
            ip_address=ip_address,
            target_object=om.id,
            event=AuditLogEntryEvent.MEMBER_INVITE,
            data=om.get_audit_log_data(),
        )

        om.send_invite_email()

        return om, True

def rollback():
    """Context manager that always rolls back to a savepoint.

    This is useful when using hypothesis (https://hypothesis.readthedocs.org/)
    which repeatedly runs tests to discover edge cases.
    """
    sid = transaction.savepoint()
    try:
        yield
    finally:
        transaction.savepoint_rollback(sid)