On Jun 24, 1:23 pm, Michael Bayer <[email protected]> wrote:
> On Jun 24, 2010, at 12:35 PM, Torsten Landschoff wrote:
> > That's what I thought but it does not cure my problem.
> > e.raw_connect().isolation_levelis in fact None, but the rollback is not
> > done anyway. :-(
>
> its passing it through.....dont know what else we can do there
I ran into this myself today and after struggling for a few hours I
came across this thread. I then ended up creating a test case to
reproduce the problem and it suggests the problem stems from some
behaviour of transactions or of the engine.base.Connection class. I
don't quite know what to make of it yet, but I think it shows that the
effect of "passing it through" is being counteracted by something
else. I ran this on Python 2.6.5 with SA 0.6.3 on Windows.
'''Test showing unexpected behaviour with sqlalchemy and sqlite3 and
isolation_level settings. The test creates one table outside of
a transaction (to detect potential problems with the test not
executing
properly) then creates a second table inside a transaction which it
immediately rolls back.
test01() fails basically as expected, since the sqlite3 DBAPI layer
appears to need isolation_level==None to properly roll back DDL
statements.
test02() succeeds because isolation_level==None now. This test and
test01()
both use a connection from engine.raw_connection(), which is a
sqlalchemy.pool._ConnectionFairy() object.
test03() tries again with isolation_level==None but using a
transaction
created from a connection returned by engine.connect(), which is a
sqlalchemy.engine.base.Connection() object. This test fails in spite
of the isolation_level setting.
'''
import unittest
from sqlalchemy import create_engine
DBPATH = 'sqlite://'
DDL = 'create table %s (id integer primary key)'
class TestCase(unittest.TestCase):
def setup(self, **kwargs):
self.engine = create_engine(DBPATH, **kwargs)
self.engine.execute(DDL % 'foo')
def rollback_raw(self):
conn = self.engine.raw_connection()
try:
conn.execute('begin')
conn.execute(DDL % 'bar')
raise ValueError
except ValueError:
conn.rollback()
def rollback_txn(self):
conn = self.engine.connect()
try:
txn = conn.begin()
conn.execute(DDL % 'bar')
raise ValueError
except ValueError:
txn.rollback()
def get_table_names(self):
conn = self.engine.raw_connection()
return [x[0] for x in conn.execute('select name from
sqlite_master')]
def test01(self):
'''use engine with default isolation_level'''
self.setup()
self.rollback_raw()
self.assertEqual(['foo'], self.get_table_names())
def test02(self):
'''use raw_connection with isolation_level None'''
self.setup(connect_args={'isolation_level': None})
self.rollback_raw()
self.assertEqual(['foo'], self.get_table_names())
def test03(self):
'''use transaction with isolation_level None'''
self.setup(connect_args={'isolation_level': None})
self.rollback_txn()
self.assertEqual(['foo'], self.get_table_names())
if __name__ == '__main__':
unittest.main()
--
Peter Hansen
Engenuity Corporation
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to [email protected].
To unsubscribe from this group, send email to
[email protected].
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
