heres how im running it:
z-eeks-Computer:~/dev/sqlalchemy classic$ export PYTHONPATH=./lib/:./
test/
z-eeks-Computer:~/dev/sqlalchemy classic$ python threadlocal-
association.py
testdelete (__main__.AssociationTest) ... ok
testinsert (__main__.AssociationTest) ... ok
testmodify (__main__.AssociationTest) ... ok
testreplace (__main__.AssociationTest) ... ok
----------------------------------------------------------------------
Ran 4 tests in 0.353s
OK
z-eeks-Computer:~/dev/sqlalchemy classic$
or postgres:
z-eeks-Computer:~/dev/sqlalchemy classic$ python threadlocal-
association.py --db postgres
testdelete (__main__.AssociationTest) ... ok
testinsert (__main__.AssociationTest) ... ok
testmodify (__main__.AssociationTest) ... ok
testreplace (__main__.AssociationTest) ... ok
----------------------------------------------------------------------
Ran 4 tests in 1.488s
OK
if you run with --verbose, that will show all the SQL and print
statements, maybe you can get a better view of whats going wrong.
script is attached....
import testbase
from sqlalchemy import *
class AssociationTest(testbase.PersistTest):
def setUpAll(self):
self.install_threadlocal()
global items, item_keywords, keywords, metadata, Item, Keyword, KeywordAssociation
metadata = BoundMetaData(testbase.db)
items = Table('items', metadata,
Column('item_id', Integer, primary_key=True),
Column('name', String(40)),
)
item_keywords = Table('item_keywords', metadata,
Column('item_id', Integer, ForeignKey('items.item_id')),
Column('keyword_id', Integer, ForeignKey('keywords.keyword_id')),
Column('data', String(40))
)
keywords = Table('keywords', metadata,
Column('keyword_id', Integer, primary_key=True),
Column('name', String(40))
)
metadata.create_all()
class Item(object):
def __init__(self, name):
self.name = name
def __repr__(self):
return "Item id=%d name=%s keywordassoc=%s" % (self.item_id, self.name, repr(self.keywords))
class Keyword(object):
def __init__(self, name):
self.name = name
def __repr__(self):
return "Keyword id=%d name=%s" % (self.keyword_id, self.name)
class KeywordAssociation(object):
def __init__(self, keyword, data):
self.keyword = keyword
self.data = data
def __repr__(self):
return "KeywordAssociation itemid=%d keyword=%s data=%s" % (self.item_id, repr(self.keyword), self.data)
mapper(Keyword, keywords)
mapper(KeywordAssociation, item_keywords, properties={
'keyword':relation(Keyword, lazy=False)
}, primary_key=[item_keywords.c.item_id, item_keywords.c.keyword_id], order_by=[item_keywords.c.data])
mapper(Item, items, properties={
'keywords' : relation(KeywordAssociation, association=Keyword)
})
def tearDown(self):
for t in metadata.table_iterator(reverse=True):
t.delete().execute()
def tearDownAll(self):
clear_mappers()
metadata.drop_all()
def testinsert(self):
item1 = Item('item1')
item2 = Item('item2')
item1.keywords.append(KeywordAssociation(Keyword('blue'), 'blue_assoc'))
item1.keywords.append(KeywordAssociation(Keyword('red'), 'red_assoc'))
item2.keywords.append(KeywordAssociation(Keyword('green'), 'green_assoc'))
objectstore.save(item2)
objectstore.flush()
saved = repr([item1, item2])
objectstore.clear()
l = objectstore.query(Item).select()
# l = Item.select()
loaded = repr(l)
print saved
print loaded
self.assert_(saved == loaded)
def testreplace(self):
item1 = Item('item1')
item1.keywords.append(KeywordAssociation(Keyword('blue'), 'blue_assoc'))
item1.keywords.append(KeywordAssociation(Keyword('red'), 'red_assoc'))
# sess.save(item1)
objectstore.flush()
red_keyword = item1.keywords[1].keyword
del item1.keywords[1]
item1.keywords.append(KeywordAssociation(red_keyword, 'new_red_assoc'))
objectstore.flush()
saved = repr([item1])
objectstore.clear()
l = objectstore.query(Item).select()
# l = Item.select()
loaded = repr(l)
print saved
print loaded
self.assert_(saved == loaded)
def testmodify(self):
item1 = Item('item1')
item2 = Item('item2')
item1.keywords.append(KeywordAssociation(Keyword('blue'), 'blue_assoc'))
item1.keywords.append(KeywordAssociation(Keyword('red'), 'red_assoc'))
item2.keywords.append(KeywordAssociation(Keyword('green'), 'green_assoc'))
# sess.save(item1)
# sess.save(item2)
objectstore.flush()
red_keyword = item1.keywords[1].keyword
del item1.keywords[0]
del item1.keywords[0]
purple_keyword = Keyword('purple')
item1.keywords.append(KeywordAssociation(red_keyword, 'new_red_assoc'))
item2.keywords.append(KeywordAssociation(purple_keyword, 'purple_item2_assoc'))
item1.keywords.append(KeywordAssociation(purple_keyword, 'purple_item1_assoc'))
item1.keywords.append(KeywordAssociation(Keyword('yellow'), 'yellow_assoc'))
objectstore.flush()
saved = repr([item1, item2])
objectstore.clear()
l = objectstore.query(Item).select()
loaded = repr(l)
print saved
print loaded
self.assert_(saved == loaded)
def testdelete(self):
item1 = Item('item1')
item2 = Item('item2')
item1.keywords.append(KeywordAssociation(Keyword('blue'), 'blue_assoc'))
item1.keywords.append(KeywordAssociation(Keyword('red'), 'red_assoc'))
item2.keywords.append(KeywordAssociation(Keyword('green'), 'green_assoc'))
# sess.save(item1)
# sess.save(item2)
objectstore.flush()
self.assert_(item_keywords.count().scalar() == 3)
objectstore.delete(item1)
objectstore.delete(item2)
objectstore.flush()
self.assert_(item_keywords.count().scalar() == 0)
if __name__ == "__main__":
testbase.main()
On Jun 28, 2006, at 3:24 AM, Raul Garcia wrote:
Michael Bayer [mailto:[EMAIL PROTECTED]
heres a diff. the test framework has the "threadlocal" thing turned
off by default and also plays some games with it.
you should be able to remove all the "save()" statements from the
tests and everything should keep working; with threadlocal running,
new objects automatically get added to the session.
All right, I've done that and the tests keep failing. I've run the
patched
test module against sqlite and postgres (on my windows setup) but
to no
avail.
also, when you say Item.select(), it looks like youre looking for
"assign_mapper()" functionality, which tacks on a bunch of
methods to
each class that automatically talk to its mapper; this is something
of a SQLObject-compatibility hack. in that case, use assign_mapper
(class, table) instead of mapper(class, table).
Yes, thanks a lot. The real code is full of selects like this (from
the
SQLObject model) and it seems that the threadlocal is the fastest way
to convert. Also, the model has to be used into a threaded CherryPy
application, so the threadlocal strategy seems the best one for this
case (one thread per request, one default connection/session per
thread).
Regards,
Raúl García.
Using Tomcat but need to do more? Need to support web services,
security?
Get stuff done quickly with pre-integrated technology to make your
job easier
Download IBM WebSphere Application Server v.1.0.1 based on Apache
Geronimo
http://sel.as-us.falkag.net/sel?
cmd=lnk&kid=120709&bid=263057&dat=121642
_______________________________________________
Sqlalchemy-users mailing list
Sqlalchemy-users@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/sqlalchemy-users
Using Tomcat but need to do more? Need to support web services, security?
Get stuff done quickly with pre-integrated technology to make your job easier
Download IBM WebSphere Application Server v.1.0.1 based on Apache Geronimo
http://sel.as-us.falkag.net/sel?cmd=lnk&kid=120709&bid=263057&dat=121642
_______________________________________________
Sqlalchemy-users mailing list
Sqlalchemy-users@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/sqlalchemy-users
