# when the task from "mapper" executes, take the objects from the task corresponding
# to "mapperfrom"'s list of save/delete objects, and send them to "processor"
# for dependency processing
- #print "registerprocessor", str(mapper), repr(processor.key), str(mapperfrom)
+
+ #print "registerprocessor", str(mapper), repr(processor), repr(processor.key), str(mapperfrom)
# correct for primary mapper (the mapper offcially associated with the class)
mapper = mapper.primary_mapper()
for taskelement in task.get_elements(polymorphic=False):
obj = taskelement.obj
object_to_original_task[obj] = task
- #print "OBJ", repr(obj), "TASK", repr(task)
for dep in deps_by_targettask.get(task, []):
# is this dependency involved in one of the cycles ?
- #print "DEP iterate", dep.processor.key, dep.processor.parent, dep.processor.mapper
if not dependency_in_cycles(dep):
- #print "NOT IN CYCLE"
continue
- #print "DEP", dep.processor.key
(processor, targettask) = (dep.processor, dep.targettask)
isdelete = taskelement.isdelete
childlist = childlist.added_items() + childlist.unchanged_items() + childlist.deleted_items()
for o in childlist:
- if o is None or not childtask.contains_object(o, polymorphic=True):
+
+ # other object is None. this can occur if the relationship is many-to-one
+ # or one-to-one, and None was set. the "removed" object will be picked
+ # up in this iteration via the deleted_items() part of the collection.
+ if o is None:
continue
- #print "parent/child", obj, o
+
+ # the other object is not in the UOWTransaction ! but if we are many-to-one,
+ # we need a task in order to attach dependency operations, so establish a "listonly"
+ # task
+ if not childtask.contains_object(o, polymorphic=True):
+ childtask.append(o, listonly=True)
+ object_to_original_task[o] = childtask
+
+ # create a tuple representing the "parent/child"
whosdep = dep.whose_dependent_on_who(obj, o)
- #print "WHOSEDEP", dep.processor.key, dep.processor.direction, whosdep
if whosdep is not None:
+ # append the tuple to the partial ordering.
tuples.append(whosdep)
+
# create a UOWDependencyProcessor representing this pair of objects.
# append it to a UOWTask
if whosdep[0] is obj:
# create a tree of UOWTasks corresponding to the tree of object instances
# created by the DependencySorter
def make_task_tree(node, parenttask, nexttasks):
- #print "MAKETASKTREE", node.item, parenttask
originating_task = object_to_original_task[node.item]
t = nexttasks.get(originating_task, None)
if t is None:
from tables import *
import tables
-# TODO: need assertion conditions in this suite
+"""test cyclical mapper relationships. Many of the assertions are provided
+via running with postgres, which is strict about foreign keys."""
-
-"""test cyclical mapper relationships. No assertions yet, but run it with postgres and the
-foreign key checks alone will usually not work if something is wrong"""
class Tester(object):
def __init__(self, data=None):
self.data = data
sess.flush()
sess.delete(a)
sess.flush()
+
+ def testmanytooneonly(self):
+ """test that the circular dependency sort can assemble a many-to-one dependency processor
+ when only the object on the "many" side is actually in the list of modified objects.
+ this requires that the circular sort add the other side of the relation into the UOWTransaction
+ so that the dependency operation can be tacked onto it.
+
+ This also affects inheritance relationships since they rely upon circular sort as well.
+ """
+ class C1(Tester):
+ pass
+ mapper(C1, t1, properties={
+ 'parent':relation(C1, primaryjoin=t1.c.parent_c1==t1.c.c1, foreignkey=t1.c.c1)
+ })
+ sess = create_session()
+ c1 = C1()
+ sess.save(c1)
+ sess.flush()
+ sess.clear()
+ c1 = sess.query(C1).get(c1.c1)
+ c2 = C1()
+ c2.parent = c1
+ sess.save(c2)
+ sess.flush()
+ assert c2.parent_c1==c1.c1
def testcycle(self):
class C1(Tester):
assert False
except exceptions.ArgumentError:
assert True
+
+class InheritTestOne(AssertMixin):
+ def setUpAll(self):
+ global parent, child1, child2, meta
+ meta = BoundMetaData(testbase.db)
+ parent = Table("parent", meta,
+ Column("id", Integer, primary_key=True),
+ Column("parent_data", String(50)),
+ Column("type", String(10))
+ )
+
+ child1 = Table("child1", meta,
+ Column("id", Integer, ForeignKey("parent.id"), primary_key=True),
+ Column("child1_data", String(50))
+ )
+
+ child2 = Table("child2", meta,
+ Column("id", Integer, ForeignKey("parent.id"), primary_key=True),
+ Column("child1_id", Integer, ForeignKey("child1.id"), nullable=False),
+ Column("child2_data", String(50))
+ )
+ meta.create_all()
+ def tearDownAll(self):
+ meta.drop_all()
+ def testmanytooneonly(self):
+ """test similar to SelfReferentialTest.testmanytooneonly"""
+ class Parent(object):
+ pass
+
+ mapper(Parent, parent)
+
+ class Child1(Parent):
+ pass
+
+ mapper(Child1, child1, inherits=Parent)
+
+ class Child2(Parent):
+ pass
+
+ mapper(Child2, child2, properties={
+ "child1": relation(Child1,
+ primaryjoin=child2.c.child1_id==child1.c.id,
+ )
+ },inherits=Parent)
+
+ session = create_session()
+
+ c1 = Child1()
+ c1.child1_data = "qwerty"
+ session.save(c1)
+ session.flush()
+ session.clear()
+
+ c1 = session.query(Child1).get_by(child1_data="qwerty")
+ c2 = Child2()
+ c2.child1 = c1
+ c2.child2_data = "asdfgh"
+ session.save(c2)
+ # the flush will fail if the UOW does not set up a many-to-one DP
+ # attached to a task corresponding to c1, since "child1_id" is not nullable
+ session.flush()
+
class BiDirectionalOneToManyTest(AssertMixin):
"""tests two mappers with a one-to-many relation to each other."""
def setUpAll(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
