together to allow processing of scalar- and list-based dependencies at flush time."""
from sync import ONETOMANY,MANYTOONE,MANYTOMANY
-from sqlalchemy import sql
+from sqlalchemy import sql, util
def create_dependency_processor(key, syncrules, cascade, secondary=None, association=None, is_backref=False, post_update=False):
types = {
uowcommit.register_processor(stub, self, self.parent)
def process_dependencies(self, task, deplist, uowcommit, delete = False):
#print self.mapper.table.name + " " + self.key + " " + repr(len(deplist)) + " process_dep isdelete " + repr(delete) + " direction " + repr(self.direction)
- # manage association objects.
for obj in deplist:
childlist = self.get_object_dependencies(obj, uowcommit, passive=True)
if childlist is None: continue
- #print "DIRECTION", self.direction
- d = {}
- for child in childlist:
- self._synchronize(obj, child, None, False)
- key = self.mapper.instance_key(child)
- #print "SYNCHRONIZED", child, "INSTANCE KEY", key
- d[key] = child
- uowcommit.unregister_object(child)
-
- for child in childlist.added_items():
- uowcommit.register_object(child)
- key = self.mapper.instance_key(child)
- #print "ADDED, INSTANCE KEY", key
- d[key] = child
-
- for child in childlist.unchanged_items():
- key = self.mapper.instance_key(child)
- o = d[key]
- o._instance_key= key
+ # for the association mapper, the list of association objects is organized into a unique list based on the
+ # "primary key". newly added association items which correspond to existing association items are "merged"
+ # into the existing one by moving the "_instance_key" over to the added item, so instead of insert/delete you
+ # just get an update operation.
+ if not delete:
+ tosave = util.OrderedDict()
+ for child in childlist:
+ self._synchronize(obj, child, None, False)
+ key = self.mapper.instance_key(child)
+ tosave[key] = child
+ uowcommit.unregister_object(child)
- for child in childlist.deleted_items():
- key = self.mapper.instance_key(child)
- #print "DELETED, INSTANCE KEY", key
- if d.has_key(key):
- o = d[key]
- o._instance_key = key
+ todelete = {}
+ for child in childlist.deleted_items():
+ self._synchronize(obj, child, None, False)
+ key = self.mapper.instance_key(child)
+ if not tosave.has_key(key):
+ todelete[key] = child
+ else:
+ tosave[key]._instance_key = key
uowcommit.unregister_object(child)
- else:
- #print "DELETE ASSOC OBJ", repr(child)
- uowcommit.register_object(child, isdelete=True)
+
+ for child in childlist.unchanged_items():
+ key = self.mapper.instance_key(child)
+ tosave[key]._instance_key = key
+
+ #print "OK for the save", [(o, getattr(o, '_instance_key', None)) for o in tosave.values()]
+ #print "OK for the delete", [(o, getattr(o, '_instance_key', None)) for o in todelete.values()]
+
+ for obj in tosave.values():
+ uowcommit.register_object(obj)
+ for obj in todelete.values():
+ uowcommit.register_object(obj, isdelete=True)
+ else:
+ todelete = {}
+ for child in childlist.unchanged_items() + childlist.deleted_items():
+ self._synchronize(obj, child, None, False)
+ key = self.mapper.instance_key(child)
+ todelete[key] = child
+ for obj in todelete.values():
+ uowcommit.register_object(obj, isdelete=True)
+
+
def preprocess_dependencies(self, task, deplist, uowcommit, delete = False):
# TODO: clean up the association step in process_dependencies and move the
# appropriate sections of it to here
"""called by a UnitOfWork object to delete objects, which involves a
DELETE statement for each table used by this mapper, for each object in the list."""
connection = uow.transaction.connection(self)
+ #print "DELETE_OBJ MAPPER", self.class_.__name__, objects
for table in util.reversed(self.tables):
if not self._has_pks(table):
if mod: self._mark_modified()
def unregister_object(self, obj):
+ #print "UNREGISTER", obj
mapper = object_mapper(obj)
task = self.get_task_by_mapper(mapper)
if obj in task.objects:
"""dumps out a string representation of a UOWTask structure"""
class UOWDumper(object):
- def __init__(self, task, buf, verbose=False):
+ def __init__(self, task, buf, verbose=True):
self.verbose = verbose
self.indent = 0
self.task = task
'objectstore',
'cascade',
'relationships',
+ 'association',
# cyclical ORM persistence
'cycles',
'manytomany',
'onetoone',
'inheritance',
- 'polymorph',
+ 'polymorph',
# extensions
'proxy_engine',
--- /dev/null
+import testbase
+
+from sqlalchemy import *
+
+
+class AssociationTest(testbase.PersistTest):
+ def setUpAll(self):
+ 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):
+ sess = create_session()
+ 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)
+ sess.flush()
+ saved = repr([item1, item2])
+ sess.clear()
+ l = sess.query(Item).select()
+ loaded = repr(l)
+ print saved
+ print loaded
+ self.assert_(saved == loaded)
+
+ def testreplace(self):
+ sess = create_session()
+ item1 = Item('item1')
+ item1.keywords.append(KeywordAssociation(Keyword('blue'), 'blue_assoc'))
+ item1.keywords.append(KeywordAssociation(Keyword('red'), 'red_assoc'))
+ sess.save(item1)
+ sess.flush()
+
+ red_keyword = item1.keywords[1].keyword
+ del item1.keywords[1]
+ item1.keywords.append(KeywordAssociation(red_keyword, 'new_red_assoc'))
+ sess.flush()
+ saved = repr([item1])
+ sess.clear()
+ l = sess.query(Item).select()
+ loaded = repr(l)
+ print saved
+ print loaded
+ self.assert_(saved == loaded)
+
+ def testmodify(self):
+ sess = create_session()
+ 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)
+ sess.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'))
+
+ sess.flush()
+ saved = repr([item1, item2])
+ sess.clear()
+ l = sess.query(Item).select()
+ loaded = repr(l)
+ print saved
+ print loaded
+ self.assert_(saved == loaded)
+
+ def testdelete(self):
+ sess = create_session()
+ 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)
+ sess.flush()
+ self.assert_(item_keywords.count().scalar() == 3)
+
+ sess.delete(item1)
+ sess.delete(item2)
+ sess.flush()
+ self.assert_(item_keywords.count().scalar() == 0)
+
+
+if __name__ == "__main__":
+ testbase.main()
-"""Test complex relationships"""
-
import testbase
import unittest, sys, datetime
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
