diff options
| -rw-r--r-- | CHANGES | 2 | ||||
| -rw-r--r-- | lib/sqlalchemy/engine/threadlocal.py | 10 | ||||
| -rw-r--r-- | test/engine/transaction.py | 32 |
3 files changed, 40 insertions, 4 deletions
@@ -1,6 +1,8 @@ 0.2.4 - try/except when the mapper sets init.__name__ on a mapped class, supports python 2.3 +- fixed bug where threadlocal engine would still autocommit +despite a transaction in progress - lazy load and deferred load operations require the parent object to be in a Session to do the operation; whereas before the operation would just return a blank list or None, it now raises an exception. diff --git a/lib/sqlalchemy/engine/threadlocal.py b/lib/sqlalchemy/engine/threadlocal.py index 913c6425a..d69a40755 100644 --- a/lib/sqlalchemy/engine/threadlocal.py +++ b/lib/sqlalchemy/engine/threadlocal.py @@ -23,10 +23,12 @@ class TLSession(object): except AttributeError: pass self.__tcount = 0 - - def begin(self): + def begin(self, tlconnection=None): if self.__tcount == 0: - self.__transaction = self.get_connection() + if tlconnection is None: + self.__transaction = self.get_connection() + else: + self.__transaction = tlconnection self.__trans = self.__transaction._begin() self.__tcount += 1 return self.__trans @@ -61,7 +63,7 @@ class TLConnection(base.Connection): def _begin(self): return base.Connection.begin(self) def begin(self): - return self.session.begin() + return self.session.begin(self) def close(self): if self.__opencount == 1: base.Connection.close(self) diff --git a/test/engine/transaction.py b/test/engine/transaction.py index b8f8af96e..5f429761e 100644 --- a/test/engine/transaction.py +++ b/test/engine/transaction.py @@ -157,6 +157,38 @@ class TLTransactionTest(testbase.PersistTest): finally: external_connection.close() + @testbase.unsupported('mysql') + def testrollback_off_conn(self): + conn = tlengine.contextual_connect() + trans = conn.begin() + conn.execute(users.insert(), user_id=1, user_name='user1') + conn.execute(users.insert(), user_id=2, user_name='user2') + conn.execute(users.insert(), user_id=3, user_name='user3') + trans.rollback() + + external_connection = tlengine.connect() + result = external_connection.execute("select * from query_users") + try: + assert len(result.fetchall()) == 0 + finally: + external_connection.close() + + @testbase.unsupported('mysql') + def testcommit_off_conn(self): + conn = tlengine.contextual_connect() + trans = conn.begin() + conn.execute(users.insert(), user_id=1, user_name='user1') + conn.execute(users.insert(), user_id=2, user_name='user2') + conn.execute(users.insert(), user_id=3, user_name='user3') + trans.commit() + + external_connection = tlengine.connect() + result = external_connection.execute("select * from query_users") + try: + assert len(result.fetchall()) == 3 + finally: + external_connection.close() + @testbase.unsupported('mysql', 'sqlite') def testnesting(self): """tests nesting of tranacstions""" |