Python transaction 模块，begin() 实例源码

def remote_create(self, command_line):
        try:
            self.callx('create', command_line)
        except CalledProcessError as cpe:
            if cpe.returncode == 1:
                log.debug('remote create finished (warning)')
                with transaction.manager as txn:
                    self.job.borg_warning = True
                    txn.note('Set borg warning flag on job %s' % self.job.id)
            else:
                raise
        else:
            log.debug('remote create finished (success)')
        finally:
            transaction.begin()
        self.job.update_state(BackupJob.State.client_in_progress, BackupJob.State.client_done)

def process_oils(session_class):
    session = session_class()
    record_ids = [r.adios_oil_id for r in session.query(ImportedRecord)]
    session.close()

    logger.info('Adding Oil objects...')
    for record_id in record_ids:
        # Note: committing our transaction for every record slows the
        #       import job significantly.  But this is necessary if we
        #       want the option of rejecting oil records.
        session = session_class()
        transaction.begin()
        rec = (session.query(ImportedRecord)
               .filter(ImportedRecord.adios_oil_id == record_id)
               .one())

        try:
            add_oil(rec)
            transaction.commit()
        except OilRejected as e:
            logger.warning(repr(e))
            transaction.abort()

def transaction(self, principal_id):
        if principal_id:
            transaction.begin()
            login_principal(get_principal(principal_id))
        try:
            yield
        except:
            transaction.abort()
            raise
        else:
            try:
                transaction.commit()
            except ZODB.POSException.ConflictError:
                log.warning('Conflict while publishing', exc_info=True)
                transaction.abort()
                raise
        finally:
            zope.security.management.endInteraction()

def test_transaction_record_rollback(session):
    import transaction
    import uuid
    from snovault.storage import Resource
    rid = uuid.uuid4()
    resource = Resource('test_item', {'': {}}, rid=rid)
    session.add(resource)
    transaction.commit()
    transaction.begin()
    sp = session.begin_nested()
    resource = Resource('test_item', {'': {}}, rid=rid)
    session.add(resource)
    with pytest.raises(Exception):
        sp.commit()
    sp.rollback()
    resource = Resource('test_item', {'': {}})
    session.add(resource)
    transaction.commit()

def conn(engine_url):
    from snovault.app import configure_engine
    from snovault.storage import Base

    engine_settings = {
        'sqlalchemy.url': engine_url,
    }

    engine = configure_engine(engine_settings)
    conn = engine.connect()
    tx = conn.begin()
    try:
        Base.metadata.create_all(bind=conn)
        yield conn
    finally:
        tx.rollback()
        conn.close()
        engine.dispose()

def idle(self):
        transaction.begin()
        hook.borgcubed_idle(apiserver=self)
        self.check_children()
        self.queue_new_jobs()
        self.check_queue()

def set_geolocation(cls, company_id, lat, lon):
        transaction.begin()
        DBSession.query(cls) \
            .filter(cls.id == company_id) \
            .update({'latitude': lat,
                     'longitude': lon,
                     'geolocation_is_valid': True})
        transaction.commit()

def set_geolocation_is_valid(cls, company_id, is_valid):
        transaction.begin()
        DBSession.query(cls) \
            .filter(cls.id == company_id) \
            .update({'geolocation_is_valid': is_valid})
        transaction.commit()

def set_address_is_valid(cls, company_id, is_valid):
        transaction.begin()
        DBSession.query(cls) \
            .filter(cls.id == company_id) \
            .update({'address_is_valid': is_valid})
        transaction.commit()

def update_last_sync(cls, job_id, timestamp):
        transaction.begin()
        DBSession.query(cls) \
            .filter(cls.id == job_id) \
            .update({'last_sync': timestamp,
                     'last_modified': cls.last_modified})
        DBSession.query(cls) \
            .filter(cls.id == job_id) \
            .filter(cls.last_modified < timestamp) \
            .update({'last_modified': timestamp})
        transaction.commit()

def reset_last_sync(cls):
        transaction.begin()
        DBSession.query(cls) \
            .filter(cls.validated) \
            .update({'last_sync': base_time()})
        transaction.commit()

def set_geolocation(cls, offer_id, lat, lon):
        transaction.begin()
        DBSession.query(cls) \
            .filter(cls.id == offer_id) \
            .update({'latitude': lat,
                     'longitude': lon,
                     'geolocation_is_valid': True})
        transaction.commit()

def set_geolocation_is_valid(cls, offer_id, is_valid):
        transaction.begin()
        DBSession.query(cls) \
            .filter(cls.id == offer_id) \
            .update({'geolocation_is_valid': is_valid})
        transaction.commit()

def set_pushed_on_twitter(cls, offer_id, pushed_on_twitter):
        transaction.begin()
        DBSession.query(cls) \
            .filter(cls.id == offer_id) \
            .update({'pushed_on_twitter': pushed_on_twitter})
        transaction.commit()

def update_last_sync(cls, job_id, timestamp):
        transaction.begin()
        DBSession.query(cls) \
            .filter(cls.id == job_id) \
            .update({'last_sync': timestamp,
                     'last_modified': cls.last_modified})
        DBSession.query(cls) \
            .filter(cls.id == job_id) \
            .filter(cls.last_modified < timestamp) \
            .update({'last_modified': timestamp})
        transaction.commit()

def reset_last_sync(cls):
        transaction.begin()
        DBSession.query(cls) \
            .update({'last_sync': base_time()})
        transaction.commit()

def automatic_transaction_begin():
    """Starts a new transaction for every test.

    We want to start with an empty celery_session for each test.

    """
    transaction.begin()
    zope.security.management.endInteraction()

def begin(self):
        transaction.begin()

def transaction(request, external_tx, zsa_savepoints, check_constraints):
    import transaction
    transaction.begin()
    request.addfinalizer(transaction.abort)
    return transaction

def set_snapshot(xmin, snapshot_id):
    global current_xmin_snapshot_id
    if current_xmin_snapshot_id == (xmin, snapshot_id):
        return
    clear_snapshot()
    current_xmin_snapshot_id = (xmin, snapshot_id)

    while True:
        txn = transaction.begin()
        txn.doom()
        if snapshot_id is not None:
            txn.setExtendedInfo('snapshot_id', snapshot_id)
        session = app.registry[DBSESSION]()
        connection = session.connection()
        db_xmin = connection.execute(
            "SELECT txid_snapshot_xmin(txid_current_snapshot());").scalar()
        if db_xmin >= xmin:
            break
        transaction.abort()
        log.info('Waiting for xmin %r to reach %r', db_xmin, xmin)
        time.sleep(0.1)

    registry = app.registry
    request = app.request_factory.blank('/_indexing_pool')
    request.registry = registry
    request.datastore = 'database'
    apply_request_extensions(request)
    request.invoke_subrequest = app.invoke_subrequest
    request.root = app.root_factory(request)
    request._stats = {}
    manager.push({'request': request, 'registry': registry})

def zsa_savepoints(conn):
    """ Place a savepoint at the start of the zope transaction

    This means failed requests rollback to the db state when they began rather
    than that at the start of the test.
    """
    from transaction.interfaces import ISynchronizer
    from zope.interface import implementer

    @implementer(ISynchronizer)
    class Savepoints(object):
        def __init__(self, conn):
            self.conn = conn
            self.sp = None
            self.state = None

        def beforeCompletion(self, transaction):
            pass

        def afterCompletion(self, transaction):
            # txn be aborted a second time in manager.begin()
            if self.sp is None:
                return
            if self.state == 'commit':
                self.state = 'completion'
                self.sp.commit()
            else:
                self.state = 'abort'
                self.sp.rollback()
            self.sp = None
            self.state = 'done'

        def newTransaction(self, transaction):
            self.state = 'new'
            self.sp = self.conn.begin_nested()
            self.state = 'begun'
            transaction.addBeforeCommitHook(self._registerCommit)

        def _registerCommit(self):
            self.state = 'commit'

    zsa_savepoints = Savepoints(conn)

    import transaction
    transaction.manager.registerSynch(zsa_savepoints)

    yield zsa_savepoints
    transaction.manager.unregisterSynch(zsa_savepoints)