Hi Group,
I would like to use SqlAlchemy QueuePool to manage some recyclable
objects.
I would like also provide my listener to QueuePool, to perform some
actions at checkin and checkout events.
Into the example below, the recyclable object is a cx_Oracle
connection. I do not need to recycle cx_Oracle connections in my
application, but here I use cx_Oracle connections just to keep the
example simple.
I run the test, but it does not behave as expected: in particular,
"connect()" of "MyListener" prints the message as expected, but
"checkout()" and "checkin()" do not.
Could you please help me and show my mistake?
I'm using SqlAlchemy 0.5.6 on Python26.
Thank you very much,
Francesca Leon
---------------------------------------
from sqlalchemy.pool import QueuePool
import cx_Oracle
import unittest
def reusableCreator():
return cx_Oracle.connect(u"username/passw...@server")
class MyListener(object):
def connect(self, dbapi_con, con_record):
print "MyListener connect"
def checkout(self, dbapi_con, con_record, con_proxy):
print "MyListener checkout"
def checkin(self, dbapi_con, con_record):
print "MyListener checkin"
class TestQueuePool(unittest.TestCase):
def setUp(self):
self.pool = QueuePool(reusableCreator)
def tearDown(self):
self.pool.dispose()
def testListener(self):
self.pool.add_listener(MyListener())
poolRecord = self.pool.get()
self.pool.return_conn(poolRecord)
--
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.
