Two Phase Transaction Support
-----------------------------
-Two Phase transactions are implemented using XA transactions. Success has been reported
-with this feature but it should be regarded as experimental.
+Two Phase transactions are implemented using XA transactions, and are known
+to work in a rudimental fashion with recent versions of cx_Oracle
+as of SQLAlchemy 0.8.0b2, 0.7.10. However, the mechanism is not yet
+considered to be robust and should still be regarded as experimental.
+
+In particular, the cx_Oracle DBAPI as recently as 5.1.2 has a bug regarding
+two phase which prevents
+a particular DBAPI connection from being consistently usable in both
+prepared transactions as well as traditional DBAPI usage patterns; therefore
+once a particular connection is used via :meth:`.Connection.begin_prepared`,
+all subsequent usages of the underlying DBAPI connection must be within
+the context of prepared transactions.
+
+The default behavior of :class:`.Engine` is to maintain a pool of DBAPI
+connections. Therefore, due to the above glitch, a DBAPI connection that has
+been used in a two-phase operation, and is then returned to the pool, will
+not be usable in a non-two-phase context. To avoid this situation,
+the application can make one of several choices:
+
+* Disable connection pooling using :class:`.NullPool`
+
+* Ensure that the particular :class:`.Engine` in use is only used
+ for two-phase operations. A :class:`.Engine` bound to an ORM
+ :class:`.Session` which includes ``twophase=True`` will consistently
+ use the two-phase transaction style.
+
+* For ad-hoc two-phase operations without disabling pooling, the DBAPI
+ connection in use can be evicted from the connection pool using the
+ :class:`.Connection.detach` method.
+
+.. versionchanged:: 0.8.0b2,0.7.10
+ Support for cx_oracle prepared transactions has been implemented
+ and tested.
+
Precision Numerics
------------------
"""
from sqlalchemy.dialects.oracle.base import OracleCompiler, OracleDialect, \
- RESERVED_WORDS, OracleExecutionContext
+ OracleExecutionContext
from sqlalchemy.dialects.oracle import base as oracle
from sqlalchemy.engine import base
from sqlalchemy import types as sqltypes, util, exc, processors
-from datetime import datetime
import random
import collections
from sqlalchemy.util.compat import decimal
# expect encoded strings or unicodes, etc.
self.dbapi_type_map = {
self.dbapi.CLOB: oracle.CLOB(),
- self.dbapi.NCLOB:oracle.NCLOB(),
+ self.dbapi.NCLOB: oracle.NCLOB(),
self.dbapi.BLOB: oracle.BLOB(),
self.dbapi.BINARY: oracle.RAW(),
}
@classmethod
def dbapi(cls):
- import cx_Oracle
+ cx_Oracle = __import__('cx_Oracle')
return cx_Oracle
def initialize(self, connection):
connection.connection.begin(*xid)
def do_prepare_twophase(self, connection, xid):
- connection.connection.prepare()
+ result = connection.connection.prepare()
+ connection.info['cx_oracle_prepared'] = result
- def do_rollback_twophase(self, connection, xid, is_prepared=True, recover=False):
+ def do_rollback_twophase(self, connection, xid, is_prepared=True,
+ recover=False):
self.do_rollback(connection.connection)
- def do_commit_twophase(self, connection, xid, is_prepared=True, recover=False):
- self.do_commit(connection.connection)
+ def do_commit_twophase(self, connection, xid, is_prepared=True,
+ recover=False):
+ if not is_prepared:
+ self.do_commit(connection.connection)
+ else:
+ oci_prepared = connection.info['cx_oracle_prepared']
+ if oci_prepared:
+ self.do_commit(connection.connection)
def do_recover_twophase(self, connection):
- pass
+ connection.info.pop('cx_oracle_prepared', None)
dialect = OracleDialect_cx_oracle
onupdate='CASCADE'))
assert_raises(exc.SAWarning, bat.create)
+
+class TwoPhaseTest(fixtures.TablesTest):
+ """test cx_oracle two phase, which remains in a semi-broken state
+ so requires a carefully written test."""
+
+ __only_on__ = 'oracle+cx_oracle'
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('datatable', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('data', String(50))
+ )
+
+ def _connection(self):
+ conn = testing.db.connect()
+ conn.detach()
+ return conn
+
+ def _assert_data(self, rows):
+ eq_(
+ testing.db.scalar("select count(*) from datatable"),
+ rows
+ )
+ def test_twophase_prepare_false(self):
+ conn = self._connection()
+ for i in xrange(2):
+ trans = conn.begin_twophase()
+ conn.execute("select 1 from dual")
+ trans.prepare()
+ trans.commit()
+ conn.close()
+ self._assert_data(0)
+
+ def test_twophase_prepare_true(self):
+ conn = self._connection()
+ for i in xrange(2):
+ trans = conn.begin_twophase()
+ conn.execute("insert into datatable (id, data) "
+ "values (%s, 'somedata')" % i)
+ trans.prepare()
+ trans.commit()
+ conn.close()
+ self._assert_data(2)
+
+ def test_twophase_rollback(self):
+ conn = self._connection()
+ trans = conn.begin_twophase()
+ conn.execute("insert into datatable (id, data) "
+ "values (%s, 'somedata')" % 1)
+ trans.rollback()
+
+ trans = conn.begin_twophase()
+ conn.execute("insert into datatable (id, data) "
+ "values (%s, 'somedata')" % 1)
+ trans.prepare()
+ trans.commit()
+
+ conn.close()
+ self._assert_data(1)
+
+ def test_not_prepared(self):
+ conn = self._connection()
+ trans = conn.begin_twophase()
+ conn.execute("insert into datatable (id, data) "
+ "values (%s, 'somedata')" % 1)
+ trans.commit()
+ conn.close()
+ self._assert_data(1)
+
class DialectTypesTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = oracle.OracleDialect()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
