Transactions and Connection Management

Managing Transactions

A newly constructed Session may be said to be in the “begin” state.In this state, the Session has not established any connection ortransactional state with any of the Engine objects that may be associatedwith it.

The Session then receives requests to operate upon a database connection.Typically, this means it is called upon to execute SQL statements using a particularEngine, which may be via Session.query(), Session.execute(),or within a flush operation of pending data, which occurs when such state existsand Session.commit() or Session.flush() is called.

As these requests are received, each new Engine encountered is associatedwith an ongoing transactional state maintained by the Session.When the first Engine is operated upon, the Session can be saidto have left the “begin” state and entered “transactional” state. For eachEngine encountered, a Connection is associated with it,which is acquired via the Engine.contextual_connect() method. If aConnection was directly associated with the Session (see Joining a Session into an External Transaction (such as for test suites)for an example of this), it isadded to the transactional state directly.

For each Connection, the Session also maintains a Transaction object,which is acquired by calling Connection.begin() on each Connection,or if the Sessionobject has been established using the flag twophase=True, a TwoPhaseTransactionobject acquired via Connection.begin_twophase(). These transactions are all committed orrolled back corresponding to the invocation of theSession.commit() and Session.rollback() methods. A commit operation willalso call the TwoPhaseTransaction.prepare() method on all transactions if applicable.

When the transactional state is completed after a rollback or commit, the Sessionreleases all Transaction and Connection resources,and goes back to the “begin” state, whichwill again invoke new Connection and Transaction objects as newrequests to emit SQL statements are received.

The example below illustrates this lifecycle:

engine = create_engine("...")
Session = sessionmaker(bind=engine)
 
# new session.   no connections are in use.
session = Session()
try:
    # first query.  a Connection is acquired
    # from the Engine, and a Transaction
    # started.
    item1 = session.query(Item).get(1)
 
    # second query.  the same Connection/Transaction
    # are used.
    item2 = session.query(Item).get(2)
 
    # pending changes are created.
    item1.foo = 'bar'
    item2.bar = 'foo'
 
    # commit.  The pending changes above
    # are flushed via flush(), the Transaction
    # is committed, the Connection object closed
    # and discarded, the underlying DBAPI connection
    # returned to the connection pool.
    session.commit()
except:
    # on rollback, the same closure of state
    # as that of commit proceeds.
    session.rollback()
    raise
finally:
    # close the Session.  This will expunge any remaining
    # objects as well as reset any existing SessionTransaction
    # state.  Neither of these steps are usually essential.
    # However, if the commit() or rollback() itself experienced
    # an unanticipated internal failure (such as due to a mis-behaved
    # user-defined event handler), .close() will ensure that
    # invalid state is removed.
    session.close()

Using SAVEPOINT

SAVEPOINT transactions, if supported by the underlying engine, may bedelineated using the begin_nested()method:

Session = sessionmaker()
session = Session()
session.add(u1)
session.add(u2)
 
session.begin_nested() # establish a savepoint
session.add(u3)
session.rollback()  # rolls back u3, keeps u1 and u2
 
session.commit() # commits u1 and u2

begin_nested() may be called any numberof times, which will issue a new SAVEPOINT with a unique identifier for eachcall. For each begin_nested() call, acorresponding rollback() orcommit() must be issued. (But note that if the return value isused as a context manager, i.e. in a with-statement, then this rollback/commitis issued by the context manager upon exiting the context, and so should not beadded explicitly.)

When begin_nested() is called, aflush() is unconditionally issued(regardless of the autoflush setting). This is so that when arollback() occurs, the full state of thesession is expired, thus causing all subsequent attribute/instance access toreference the full state of the Session rightbefore begin_nested() was called.

begin_nested(), in the same manner as the less oftenused begin() method, returns a SessionTransaction objectwhich works as a context manager.It can be succinctly used around individual record inserts in order to catchthings like unique constraint exceptions:

for record in records:
    try:
        with session.begin_nested():
            session.merge(record)
    except:
        print("Skipped record %s" % record)
session.commit()

Autocommit Mode

The examples of session lifecycle at Managing Transactions referto a Session that runs in its default mode of autocommit=False.In this mode, the Session begins new transactions automaticallyas soon as it needs to do work upon a database connection; the transactionthen stays in progress until the Session.commit() or Session.rollback()methods are called.

The Session also features an older legacy mode of use calledautocommit mode, where a transaction is not started implicitly, and unlessthe Session.begin() method is invoked, the Session willperform each database operation on a new connection checked out from theconnection pool, which is then released back to the pool immediatelyafter the operation completes. This refers tomethods like Session.execute() as well as when executing a queryreturned by Session.query(). For a flush operation, the Sessionstarts a new transaction for the duration of the flush, and commits it whencomplete.

Warning

“autocommit” mode is a legacy mode of use and should not beconsidered for new projects. If autocommit mode is used, it is stronglyadvised that the application at least ensure that transaction scopeis made present via the Session.begin() method, rather thanusing the session in pure autocommit mode.

If the Session.begin() method is not used, and operations are allowedto proceed using ad-hoc connections with immediate autocommit, then theapplication probably should set autoflush=False, expire_on_commit=False,since these features are intended to be used only within the contextof a database transaction.

Modern usage of “autocommit mode” tends to be for framework integrations thatwish to control specifically when the “begin” state occurs. A session which isconfigured with autocommit=True may be placed into the “begin” state usingthe Session.begin() method. After the cycle completes uponSession.commit() or Session.rollback(), connection andtransaction resources are released and the Session goes backinto “autocommit” mode, until Session.begin() is called again:

Session = sessionmaker(bind=engine, autocommit=True)
session = Session()
session.begin()
try:
    item1 = session.query(Item).get(1)
    item2 = session.query(Item).get(2)
    item1.foo = 'bar'
    item2.bar = 'foo'
    session.commit()
except:
    session.rollback()
    raise

The Session.begin() method also returns a transactional token which iscompatible with the with statement:

Session = sessionmaker(bind=engine, autocommit=True)
session = Session()
with session.begin():
    item1 = session.query(Item).get(1)
    item2 = session.query(Item).get(2)
    item1.foo = 'bar'
    item2.bar = 'foo'

Using Subtransactions with Autocommit

A subtransaction indicates usage of the Session.begin() method in conjunction withthe subtransactions=True flag. This produces a non-transactional, delimiting construct thatallows nesting of calls to begin() and commit().Its purpose is to allow the construction of code that can function within a transactionboth independently of any external code that starts a transaction,as well as within a block that has already demarcated a transaction.

subtransactions=True is generally only useful in conjunction withautocommit, and is equivalent to the pattern described at Nesting of Transaction Blocks,where any number of functions can call Connection.begin() and Transaction.commit()as though they are the initiator of the transaction, but in fact may be participatingin an already ongoing transaction:

# method_a starts a transaction and calls method_b
def method_a(session):
    session.begin(subtransactions=True)
    try:
        method_b(session)
        session.commit()  # transaction is committed here
    except:
        session.rollback() # rolls back the transaction
        raise
 
# method_b also starts a transaction, but when
# called from method_a participates in the ongoing
# transaction.
def method_b(session):
    session.begin(subtransactions=True)
    try:
        session.add(SomeObject('bat', 'lala'))
        session.commit()  # transaction is not committed yet
    except:
        session.rollback() # rolls back the transaction, in this case
                           # the one that was initiated in method_a().
        raise
 
# create a Session and call method_a
session = Session(autocommit=True)
method_a(session)
session.close()

Subtransactions are used by the Session.flush() process to ensure that theflush operation takes place within a transaction, regardless of autocommit. Whenautocommit is disabled, it is still useful in that it forces the Sessioninto a “pending rollback” state, as a failed flush cannot be resumed in mid-operation,where the end user still maintains the “scope” of the transaction overall.

Enabling Two-Phase Commit

For backends which support two-phase operation (currently MySQL andPostgreSQL), the session can be instructed to use two-phase commit semantics.This will coordinate the committing of transactions across databases so thatthe transaction is either committed or rolled back in all databases. You canalso prepare() the session forinteracting with transactions not managed by SQLAlchemy. To use two phasetransactions set the flag twophase=True on the session:

engine1 = create_engine('postgresql://db1')
engine2 = create_engine('postgresql://db2')
 
Session = sessionmaker(twophase=True)
 
# bind User operations to engine 1, Account operations to engine 2
Session.configure(binds={User:engine1, Account:engine2})
 
session = Session()
 
# .... work with accounts and users
 
# commit.  session will issue a flush to all DBs, and a prepare step to all DBs,
# before committing both transactions
session.commit()

Setting Transaction Isolation Levels

Isolation refers to the behavior of the transaction at the databaselevel in relation to other transactions occurring concurrently. Thereare four well-known modes of isolation, and typically the Python DBAPIallows these to be set on a per-connection basis, either through explicitAPIs or via database-specific calls.

SQLAlchemy’s dialects support settable isolation modes on a per-Engineor per-Connection basis, using flags at both thecreate_engine() level as well as at the Connection.execution_options()level.

When using the ORM Session, it acts as a facade for engines andconnections, but does not expose transaction isolation directly. So inorder to affect transaction isolation level, we need to act upon theEngine or Connection as appropriate.

See also

create_engine.isolation_level

SQLite Transaction Isolation

PostgreSQL Isolation Level

MySQL Isolation Level

Setting Isolation Engine-Wide

To set up a Session or sessionmaker with a specificisolation level globally, use the create_engine.isolation_levelparameter:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
 
eng = create_engine(
    "postgresql://scott:tiger@localhost/test",
    isolation_level='REPEATABLE_READ')
 
maker = sessionmaker(bind=eng)
 
session = maker()

Setting Isolation for Individual Sessions

When we make a new Session, either using the constructor directlyor when we call upon the callable produced by a sessionmaker,we can pass the bind argument directly, overriding the pre-existing bind.We can combine this with the Engine.execution_options() methodin order to produce a copy of the original Engine that willadd this option:

session = maker(
    bind=engine.execution_options(isolation_level='SERIALIZABLE'))

For the case where the Session or sessionmaker isconfigured with multiple “binds”, we can either re-specify the bindsargument fully, or if we want to only replace specific binds, wecan use the Session.bind_mapper() or Session.bind_table()methods:

session = maker()
session.bind_mapper(
    User, user_engine.execution_options(isolation_level='SERIALIZABLE'))

We can also use the individual transaction method that follows.

Setting Isolation for Individual Transactions

A key caveat regarding isolation level is that the setting cannot besafely modified on a Connection where a transaction has alreadystarted. Databases cannot change the isolation level of a transactionin progress, and some DBAPIs and SQLAlchemy dialectshave inconsistent behaviors in this area. Some may implicitly emit aROLLBACK and some may implicitly emit a COMMIT, others may ignore the settinguntil the next transaction. Therefore SQLAlchemy emits a warning if thisoption is set when a transaction is already in play. The Sessionobject does not provide for us a Connection for use in a transactionwhere the transaction is not already begun. So here, we need to passexecution options to the Session at the start of a transactionby passing Session.connection.execution_optionsprovided by the Session.connection() method:

from sqlalchemy.orm import Session
 
sess = Session(bind=engine)
sess.connection(execution_options={'isolation_level': 'SERIALIZABLE'})
 
# work with session
 
# commit transaction.  the connection is released
# and reverted to its previous isolation level.
sess.commit()

Above, we first produce a Session using either the constructoror a sessionmaker. Then we explicitly set up the start ofa transaction by calling upon Session.connection(), which providesfor execution options that will be passed to the connection before thetransaction is begun. If we are working with a Session thathas multiple binds or some other custom scheme for Session.get_bind(),we can pass additional arguments to Session.connection() in order toaffect how the bind is procured:

sess = my_sesssionmaker()
 
# set up a transaction for the bind associated with
# the User mapper
sess.connection(
    mapper=User,
    execution_options={'isolation_level': 'SERIALIZABLE'})
 
# work with session
 
# commit transaction.  the connection is released
# and reverted to its previous isolation level.
sess.commit()

The Session.connection.execution_options argument is onlyaccepted on the first call to Session.connection() for aparticular bind within a transaction. If a transaction is already begunon the target connection, a warning is emitted:

>>> session = Session(eng)
>>> session.execute("select 1")
<sqlalchemy.engine.result.ResultProxy object at 0x1017a6c50>
>>> session.connection(execution_options={'isolation_level': 'SERIALIZABLE'})
sqlalchemy/orm/session.py:310: SAWarning: Connection is already established
for the given bind; execution_options ignored

New in version 0.9.9: Added theSession.connection.execution_optionsparameter to Session.connection().

Tracking Transaction State with Events

See the section Transaction Events for an overviewof the available event hooks for session transaction state changes.

Joining a Session into an External Transaction (such as for test suites)

If a Connection is being used which is already in a transactionalstate (i.e. has a Transaction established), a Session canbe made to participate within that transaction by just binding theSession to that Connection. The usual rationale for thisis a test suite that allows ORM code to work freely with a Session,including the ability to call Session.commit(), where afterwards theentire database interaction is rolled back:

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from unittest import TestCase
 
# global application scope.  create Session class, engine
Session = sessionmaker()
 
engine = create_engine('postgresql://...')
 
class SomeTest(TestCase):
    def setUp(self):
        # connect to the database
        self.connection = engine.connect()
 
        # begin a non-ORM transaction
        self.trans = self.connection.begin()
 
        # bind an individual Session to the connection
        self.session = Session(bind=self.connection)
 
    def test_something(self):
        # use the session in tests.
 
        self.session.add(Foo())
        self.session.commit()
 
    def tearDown(self):
        self.session.close()
 
        # rollback - everything that happened with the
        # Session above (including calls to commit())
        # is rolled back.
        self.trans.rollback()
 
        # return connection to the Engine
        self.connection.close()

Above, we issue Session.commit() as well asTransaction.rollback(). This is an example of where we take advantageof the Connection object’s ability to maintain subtransactions, ornested begin/commit-or-rollback pairs where only the outermost begin/commitpair actually commits the transaction, or if the outermost block rolls back,everything is rolled back.

Supporting Tests with Rollbacks

The above recipe works well for any kind of database enabled test, exceptfor a test that needs to actually invoke Session.rollback() withinthe scope of the test itself. The above recipe can be expanded, suchthat the Session always runs all operations within the scopeof a SAVEPOINT, which is established at the start of each transaction,so that tests can also rollback the “transaction” as well while stillremaining in the scope of a larger “transaction” that’s never committed,using two extra events:

from sqlalchemy import event
 
 
class SomeTest(TestCase):
 
    def setUp(self):
        # connect to the database
        self.connection = engine.connect()
 
        # begin a non-ORM transaction
        self.trans = connection.begin()
 
        # bind an individual Session to the connection
        self.session = Session(bind=self.connection)
 
        # start the session in a SAVEPOINT...
        self.session.begin_nested()
 
        # then each time that SAVEPOINT ends, reopen it
        @event.listens_for(self.session, "after_transaction_end")
        def restart_savepoint(session, transaction):
            if transaction.nested and not transaction._parent.nested:
 
                # ensure that state is expired the way
                # session.commit() at the top level normally does
                # (optional step)
                session.expire_all()
 
                session.begin_nested()
 
    # ... the tearDown() method stays the same