What is the difference between engine.contextual_connect() and
session.connection()?
My question can be better expressed with unittest. Code fails on
locations marked with HERE if I use contextual_connect(). The results
are same if I use Oracle.
# Test code:
import unittest
import pkg_resources
pkg_resources.require("SQLalchemy==0.3.0")
from sqlalchemy import *
from sqlalchemy.ext.activemapper import *
uri1 = 'sqlite:///test1.sqlite'
meta1 = BoundMetaData(uri1, echo=False, strategy='threadlocal')
session = objectstore
class Tab1(ActiveMapper):
class mapping:
__table__ = 'tab1'
__metadata__ = meta1
name = column(Unicode(100), primary_key=True)
from pysqlite2 import dbapi2 as sqlite
class TestSessionBase(unittest.TestCase):
def setUp(self):
create_tables()
session.clear()
self.con1 = sqlite.connect('test1.sqlite')
# con1 is separate connection and there should be no
# data visible in database until all work is commited (via
normal
# connection). The purpose of con1 is to independently test
commits...
def tearDown(self):
self.con1.close()
drop_tables()
def sel(self, sql, con):
c = con.cursor()
c.execute(sql)
res = c.fetchall()
c.close()
return res
class TestSessionConnection(TestSessionBase):
def test_connection_1(self):
transaction = session.create_transaction()
con1 = session.connection(Tab1)
con2 = session.connection(Tab1)
con3 = meta1.engine.contextual_connect()
self.assertEqual(con1, con2)
self.assertEqual(con1.connection, con3.connection)
transaction.commit()
def test_connection_2(self):
transaction = session.create_transaction()
con3 = meta1.engine.contextual_connect()
con1 = session.connection(Tab1)
con2 = session.connection(Tab1)
self.assertEqual(con1, con2)
self.assertEqual(con1.connection, con3.connection)
transaction.commit()
def _do_test_con(self, contextual):
transaction = session.create_transaction()
if contextual:
con = meta1.engine.contextual_connect()
else:
con = session.connection(Tab1)
tran = con.begin()
con.execute("insert into tab1 (name) values ('Plain')")
self.assertEqual(len(self.sel("select * from tab1",
con.connection)), 1)
self.assertEqual(len(self.sel("select * from tab1",
self.con1)), 0)
tran.commit()
self.assertEqual(len(self.sel("select * from tab1",
con.connection)), 1)
self.assertEqual(len(self.sel("select * from tab1",
self.con1)), 0) # HERE
transaction.rollback()
self.assertEqual(len(self.sel("select * from tab1",
self.con1)), 0) # HERE
def test_connection_3(self):
self._do_test_con(False)
def test_connection_4(self):
self._do_test_con(True)
if __name__ == '__main__':
unittest.main()
I'm not sure if the fact that I'm using threadlocal strategy and
activemapper makes any difference?
Thanks,
Tvrtko
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com
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
-~----------~----~----~----~------~----~------~--~---
