- A warning is emitted if the ``isolation_level`` parameter is used

with :meth:`.Connection.execution_options` when a :class:`.Transaction`
is in play; DBAPIs and/or SQLAlchemy dialects such as psycopg2,
MySQLdb may implicitly rollback or commit the transaction, or
not change the setting til next transaction, so this is never safe.
- Added new parameter :paramref:`.Session.connection.execution_options`
which may be used to set up execution options on a :class:`.Connection`
when it is first checked out, before the transaction has begun.
This is used to set up options such as isolation level on the
connection before the transaction starts.
- added new documentation section
detailing best practices for setting transaction isolation with
sessions.
fixes #3296

9 files changed, 279 insertions, 15 deletions

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index 99201ea01..d1495274a 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -15,6 +15,32 @@
     :version: 0.9.9
 
     .. change::
+        :tags: feature, orm
+        :tickets: 3296
+
+        Added new parameter :paramref:`.Session.connection.execution_options`
+        which may be used to set up execution options on a :class:`.Connection`
+        when it is first checked out, before the transaction has begun.
+        This is used to set up options such as isolation level on the
+        connection before the transaction starts.
+
+        .. seealso::
+
+            :ref:`session_transaction_isolation` - new documentation section
+            detailing best practices for setting transaction isolation with
+            sessions.
+
+    .. change::
+        :tags: bug, engine
+        :tickets: 3296
+
+        A warning is emitted if the ``isolation_level`` parameter is used
+        with :meth:`.Connection.execution_options` when a :class:`.Transaction`
+        is in play; DBAPIs and/or SQLAlchemy dialects such as psycopg2,
+        MySQLdb may implicitly rollback or commit the transaction, or
+        not change the setting til next transaction, so this is never safe.
+
+    .. change::
         :tags: bug, orm
         :tickets: 3300
 
diff --git a/doc/build/orm/session_transaction.rst b/doc/build/orm/session_transaction.rst
index ce5757dd0..24a844650 100644
--- a/doc/build/orm/session_transaction.rst
+++ b/doc/build/orm/session_transaction.rst
@@ -271,6 +271,149 @@ transactions set the flag ``twophase=True`` on the session::
     # before committing both transactions
     session.commit()
 
+
+.. _session_transaction_isolation:
+
+Setting Transaction Isolation Levels
+------------------------------------
+
+:term:`isolation` refers to the behavior of the transaction at the database
+level in relation to other transactions occurring concurrently.  There
+are four well-known modes of isolation, and typically the Python DBAPI
+allows these to be set on a per-connection basis, either through explicit
+APIs or via database-specific calls.
+
+SQLAlchemy's dialects support settable isolation modes on a per-:class:`.Engine`
+or per-:class:`.Connection` basis, using flags at both the
+:func:`.create_engine` level as well as at the :meth:`.Connection.execution_options`
+level.
+
+When using the ORM :class:`.Session`, it acts as a *facade* for engines and
+connections, but does not expose transaction isolation directly.  So in
+order to affect transaction isolation level, we need to act upon the
+:class:`.Engine` or :class:`.Connection` as appropriate.
+
+.. seealso::
+
+    :paramref:`.create_engine.isolation_level`
+
+    :ref:`SQLite Transaction Isolation <sqlite_isolation_level>`
+
+    :ref:`Postgresql Isolation Level <postgresql_isolation_level>`
+
+    :ref:`MySQL Isolation Level <mysql_isolation_level>`
+
+Setting Isolation Engine-Wide
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+To set up a :class:`.Session` or :class:`.sessionmaker` with a specific
+isolation level globally, use the :paramref:`.create_engine.isolation_level`
+parameter::
+
+    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 :class:`.Session`, either using the constructor directly
+or when we call upon the callable produced by a :class:`.sessionmaker`,
+we can pass the ``bind`` argument directly, overriding the pre-existing bind.
+We can combine this with the :meth:`.Engine.execution_options` method
+in order to produce a copy of the original :class:`.Engine` that will
+add this option::
+
+    session = maker(
+        bind=engine.execution_options(isolation_level='SERIALIZABLE'))
+
+For the case where the :class:`.Session` or :class:`.sessionmaker` is
+configured with multiple "binds", we can either re-specify the ``binds``
+argument fully, or if we want to only replace specific binds, we
+can use the :meth:`.Session.bind_mapper` or :meth:`.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 be
+safely modified on a :class:`.Connection` where a transaction has already
+started.  Databases cannot change the isolation level of a transaction
+in progress, and some DBAPIs and SQLAlchemy dialects
+have inconsistent behaviors in this area.  Some may implicitly emit a
+ROLLBACK and some may implicitly emit a COMMIT, others may ignore the setting
+until the next transaction.  Therefore SQLAlchemy emits a warning if this
+option is set when a transaction is already in play.  The :class:`.Session`
+object does not provide for us a :class:`.Connection` for use in a transaction
+where the transaction is not already begun.  So here, we need to pass
+execution options to the :class:`.Session` at the start of a transaction
+by passing :paramref:`.Session.connection.execution_options`
+provided by the :meth:`.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 :class:`.Session` using either the constructor
+or a :class:`.sessionmaker`.   Then we explicitly set up the start of
+a transaction by calling upon :meth:`.Session.connection`, which provides
+for execution options that will be passed to the connection before the
+transaction is begun.   If we are working with a :class:`.Session` that
+has multiple binds or some other custom scheme for :meth:`.Session.get_bind`,
+we can pass additional arguments to :meth:`.Session.connection` in order to
+affect 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 :paramref:`.Session.connection.execution_options` argument is only
+accepted on the **first** call to :meth:`.Session.connection` for a
+particular bind within a transaction.  If a transaction is already begun
+on 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
+
+.. versionadded:: 0.9.9 Added the
+    :paramref:`.Session.connection.execution_options`
+    parameter to :meth:`.Session.connection`.
+
 .. _session_external_transaction:
 
 Joining a Session into an External Transaction (such as for test suites)
diff --git a/lib/sqlalchemy/engine/__init__.py b/lib/sqlalchemy/engine/__init__.py
index f512e260a..7a14cdfb5 100644
--- a/lib/sqlalchemy/engine/__init__.py
+++ b/lib/sqlalchemy/engine/__init__.py
@@ -275,6 +275,8 @@ def create_engine(*args, **kwargs):
 
             :ref:`MySQL Transaction Isolation <mysql_isolation_level>`
 
+            :ref:`session_transaction_isolation` - for the ORM
+
     :param label_length=None: optional integer value which limits
         the size of dynamically generated column labels to that many
         characters. If less than 6, labels are generated as
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 8d816b7fd..8d6dd636a 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -222,6 +222,19 @@ class Connection(Connectable):
           is returned to the connection pool, i.e.
           the :meth:`.Connection.close` method is called.
 
+          .. warning::  The ``isolation_level`` execution option should
+             **not** be used when a transaction is already established, that
+             is, the :meth:`.Connection.begin` method or similar has been
+             called.  A database cannot change the isolation level on a
+             transaction in progress, and different DBAPIs and/or
+             SQLAlchemy dialects may implicitly roll back or commit
+             the transaction, or not affect the connection at all.
+
+             .. versionchanged:: 0.9.9 A warning is emitted when the
+                ``isolation_level`` execution option is used after a
+                transaction has been started with :meth:`.Connection.begin`
+                or similar.
+
           .. seealso::
 
                 :paramref:`.create_engine.isolation_level`
@@ -235,6 +248,7 @@ class Connection(Connectable):
 
                 :ref:`MySQL Transaction Isolation <mysql_isolation_level>`
 
+                :ref:`session_transaction_isolation` - for the ORM
 
         :param no_parameters: When ``True``, if the final parameter
           list or dictionary is totally empty, will invoke the
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index f6c2263b3..17d2e2531 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -395,6 +395,12 @@ class DefaultDialect(interfaces.Dialect):
             self._set_connection_isolation(connection, opts['isolation_level'])
 
     def _set_connection_isolation(self, connection, level):
+        if connection.in_transaction():
+            util.warn(
+                "Connection is already established with a Transaction; "
+                "setting isolation_level may implicitly rollback or commit "
+                "the existing transaction, or have no effect until "
+                "next transaction")
         self.set_isolation_level(connection.connection, level)
         connection.connection._connection_record.\
             finalize_callback.append(self.reset_isolation_level)
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 0e272dc95..3df6dce7a 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -226,10 +226,10 @@ class SessionTransaction(object):
     def _is_transaction_boundary(self):
         return self.nested or not self._parent
 
-    def connection(self, bindkey, **kwargs):
+    def connection(self, bindkey, execution_options=None, **kwargs):
         self._assert_active()
         bind = self.session.get_bind(bindkey, **kwargs)
-        return self._connection_for_bind(bind)
+        return self._connection_for_bind(bind, execution_options)
 
     def _begin(self, nested=False):
         self._assert_active()
@@ -301,14 +301,18 @@ class SessionTransaction(object):
             self._parent._deleted.update(self._deleted)
             self._parent._key_switches.update(self._key_switches)
 
-    def _connection_for_bind(self, bind):
+    def _connection_for_bind(self, bind, execution_options):
         self._assert_active()
 
         if bind in self._connections:
+            if execution_options:
+                util.warn(
+                    "Connection is already established for the "
+                    "given bind; execution_options ignored")
             return self._connections[bind][0]
 
         if self._parent:
-            conn = self._parent._connection_for_bind(bind)
+            conn = self._parent._connection_for_bind(bind, execution_options)
             if not self.nested:
                 return conn
         else:
@@ -321,6 +325,9 @@ class SessionTransaction(object):
             else:
                 conn = bind.contextual_connect()
 
+        if execution_options:
+            conn = conn.execution_options(**execution_options)
+
         if self.session.twophase and self._parent is None:
             transaction = conn.begin_twophase()
         elif self.nested:
@@ -793,6 +800,7 @@ class Session(_SessionClassMethods):
     def connection(self, mapper=None, clause=None,
                    bind=None,
                    close_with_result=False,
+                   execution_options=None,
                    **kw):
         """Return a :class:`.Connection` object corresponding to this
         :class:`.Session` object's transactional state.
@@ -837,6 +845,18 @@ class Session(_SessionClassMethods):
           configured with ``autocommit=True`` and does not already have a
           transaction in progress.
 
+        :param execution_options: a dictionary of execution options that will
+         be passed to :meth:`.Connection.execution_options`, **when the
+         connection is first procured only**.   If the connection is already
+         present within the :class:`.Session`, a warning is emitted and
+         the arguments are ignored.
+
+         .. versionadded:: 0.9.9
+
+         .. seealso::
+
+            :ref:`session_transaction_isolation`
+
         :param \**kw:
           Additional keyword arguments are sent to :meth:`get_bind()`,
           allowing additional arguments to be passed to custom
@@ -847,13 +867,18 @@ class Session(_SessionClassMethods):
             bind = self.get_bind(mapper, clause=clause, **kw)
 
         return self._connection_for_bind(bind,
-                                         close_with_result=close_with_result)
+                                         close_with_result=close_with_result,
+                                         execution_options=execution_options)
 
-    def _connection_for_bind(self, engine, **kwargs):
+    def _connection_for_bind(self, engine, execution_options=None, **kw):
         if self.transaction is not None:
-            return self.transaction._connection_for_bind(engine)
+            return self.transaction._connection_for_bind(
+                engine, execution_options)
         else:
-            return engine.contextual_connect(**kwargs)
+            conn = engine.contextual_connect(**kw)
+            if execution_options:
+                conn = conn.execution_options(**execution_options)
+            return conn
 
     def execute(self, clause, params=None, mapper=None, bind=None, **kw):
         """Execute a SQL expression construct or string statement within
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index 0f5bb4cb5..b7d900917 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -1,8 +1,6 @@
 from sqlalchemy.testing import eq_, assert_raises, \
-    assert_raises_message, ne_
+    assert_raises_message, ne_, expect_warnings
 import sys
-import time
-import threading
 from sqlalchemy import event
 from sqlalchemy.testing.engines import testing_engine
 from sqlalchemy import create_engine, MetaData, INT, VARCHAR, Sequence, \
@@ -1361,6 +1359,30 @@ class IsolationLevelTest(fixtures.TestBase):
         c3.close()
         c4.close()
 
+    def test_warning_in_transaction(self):
+        eng = testing_engine()
+        c1 = eng.connect()
+        with expect_warnings(
+            "Connection is already established with a Transaction; "
+            "setting isolation_level may implicitly rollback or commit "
+            "the existing transaction, or have no effect until next "
+            "transaction"
+        ):
+            with c1.begin():
+                c1 = c1.execution_options(
+                    isolation_level=self._non_default_isolation_level()
+                )
+
+                eq_(
+                    eng.dialect.get_isolation_level(c1.connection),
+                    self._non_default_isolation_level()
+                )
+        # stays outside of transaction
+        eq_(
+            eng.dialect.get_isolation_level(c1.connection),
+            self._non_default_isolation_level()
+        )
+
     def test_per_statement_bzzt(self):
         assert_raises_message(
             exc.ArgumentError,
diff --git a/test/orm/test_bind.py b/test/orm/test_bind.py
index 33cd66ebc..c5ddf151b 100644
--- a/test/orm/test_bind.py
+++ b/test/orm/test_bind.py
@@ -150,14 +150,14 @@ class BindIntegrationTest(_fixtures.FixtureTest):
         u = User(name='u1')
         sess.add(u)
         sess.flush()
-        assert transaction._connection_for_bind(testing.db) \
-            is transaction._connection_for_bind(c) is c
+        assert transaction._connection_for_bind(testing.db, None) \
+            is transaction._connection_for_bind(c, None) is c
 
         assert_raises_message(sa.exc.InvalidRequestError,
                               'Session already has a Connection '
                               'associated',
                               transaction._connection_for_bind,
-                              testing.db.connect())
+                              testing.db.connect(), None)
         transaction.rollback()
         assert len(sess.query(User).all()) == 0
         sess.close()
diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index 1d7e8e693..7684a0225 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -3,7 +3,7 @@ from sqlalchemy import (
     testing, exc as sa_exc, event, String, Column, Table, select, func)
 from sqlalchemy.testing import (
     fixtures, engines, eq_, assert_raises, assert_raises_message,
-    assert_warnings)
+    assert_warnings, mock, expect_warnings)
 from sqlalchemy.orm import (
     exc as orm_exc, Session, mapper, sessionmaker, create_session,
     relationship, attributes)
@@ -498,6 +498,32 @@ class SessionTransactionTest(FixtureTest):
         )
         return sess, u1
 
+    def test_execution_options_begin_transaction(self):
+        bind = mock.Mock()
+        sess = Session(bind=bind)
+        c1 = sess.connection(execution_options={'isolation_level': 'FOO'})
+        eq_(
+            bind.mock_calls,
+            [
+                mock.call.contextual_connect(),
+                mock.call.contextual_connect().
+                execution_options(isolation_level='FOO'),
+                mock.call.contextual_connect().execution_options().begin()
+            ]
+        )
+        eq_(c1, bind.contextual_connect().execution_options())
+
+    def test_execution_options_ignored_mid_transaction(self):
+        bind = mock.Mock()
+        conn = mock.Mock(engine=bind)
+        bind.contextual_connect = mock.Mock(return_value=conn)
+        sess = Session(bind=bind)
+        sess.execute("select 1")
+        with expect_warnings(
+                "Connection is already established for the "
+                "given bind; execution_options ignored"):
+            sess.connection(execution_options={'isolation_level': 'FOO'})
+
     def test_warning_on_using_inactive_session_new(self):
         User = self.classes.User