"""determines our 'direction', i.e. do we represent one to many, many to many, etc."""
#print self.key, repr(self.parent.table.name), repr(self.parent.primarytable.name), repr(self.foreignkey.table.name), repr(self.target), repr(self.foreigntable.name)
- if self.parent.table is self.target:
+ if self.secondaryjoin is not None:
+ return PropertyLoader.MANYTOMANY
+ elif self.parent.table is self.target:
if self.foreignkey.primary_key:
return PropertyLoader.MANYTOONE
else:
return PropertyLoader.ONETOMANY
- elif self.secondaryjoin is not None:
- return PropertyLoader.MANYTOMANY
elif self.foreigntable == self.mapper.noninherited_table:
return PropertyLoader.ONETOMANY
elif self.foreigntable == self.parent.noninherited_table:
binary.right = binds.setdefault(binary.right,
sql.BindParamClause(binary.left._label, None, shortname = binary.right.name))
- if secondaryjoin is not None:
- lazywhere = sql.and_(primaryjoin, secondaryjoin)
- else:
- lazywhere = primaryjoin
- lazywhere = lazywhere.copy_container()
+ lazywhere = primaryjoin.copy_container()
li = BinaryVisitor(visit_binary)
lazywhere.accept_visitor(li)
+ if secondaryjoin is not None:
+ lazywhere = sql.and_(lazywhere, secondaryjoin)
return (lazywhere, binds)
'''represents a place'''
def __init__(self, name=None):
self.name = name
+ def __str__(self):
+ return "(Place '%s')" % self.name
+ def __repr__(self):
+ return str(self)
class PlaceThingy(object):
'''represents a thingy attached to a Place'''
Column('transition_id', Integer, ForeignKey('transition.transition_id')),
)
+ global place_place
+ place_place = Table('place_place', db,
+ Column('pl1_id', Integer, ForeignKey('place.place_id')),
+ Column('pl2_id', Integer, ForeignKey('place.place_id')),
+ )
+
place.create()
transition.create()
place_input.create()
place_output.create()
place_thingy.create()
+ place_place.create()
def tearDownAll(self):
+ place_place.drop()
place_input.drop()
place_output.drop()
place_thingy.drop()
place.drop()
transition.drop()
+ #testbase.db.tables.clear()
def setUp(self):
objectstore.clear()
clear_mappers()
def tearDown(self):
+ place_place.delete().execute()
place_input.delete().execute()
place_output.delete().execute()
transition.delete().execute()
place.delete().execute()
+ def testcircular(self):
+ """tests a many-to-many relationship from a table to itself."""
+
+ Place.mapper = mapper(Place, place)
+
+ Place.mapper.add_property('places', relation(
+ Place.mapper, secondary=place_place, primaryjoin=place.c.place_id==place_place.c.pl1_id,
+ secondaryjoin=place.c.place_id==place_place.c.pl2_id,
+ order_by=place_place.c.pl2_id,
+ lazy=True,
+ ))
+
+ p1 = Place('place1')
+ p2 = Place('place2')
+ p3 = Place('place3')
+ p4 = Place('place4')
+ p5 = Place('place5')
+ p6 = Place('place6')
+ p7 = Place('place7')
+
+ p1.places.append(p2)
+ p1.places.append(p3)
+ p5.places.append(p6)
+ p6.places.append(p1)
+ p7.places.append(p1)
+ p1.places.append(p5)
+ p4.places.append(p3)
+ p3.places.append(p4)
+ objectstore.flush()
+
+ objectstore.clear()
+ l = Place.mapper.select(order_by=place.c.place_id)
+ (p1, p2, p3, p4, p5, p6, p7) = l
+ assert p1.places == [p2,p3,p5]
+ assert p5.places == [p6]
+ assert p7.places == [p1]
+ assert p6.places == [p1]
+ assert p4.places == [p3]
+ assert p3.places == [p4]
+ assert p2.places == []
+
+ for p in l:
+ pp = p.places
+ self.echo("Place " + str(p) +" places " + repr(pp))
+
+ objectstore.delete(p1,p2,p3,p4,p5,p6,p7)
+ objectstore.flush()
+
def testdouble(self):
"""tests that a mapper can have two eager relations to the same table, via
two different association tables. aliases are required."""
}
)
- def testcircular(self):
- """tests a circular many-to-many relationship. this requires that the mapper
- "break off" a new "mapper stub" to indicate a third depedendent processor."""
+ def testbidirectional(self):
+ """tests a bi-directional many-to-many relationship."""
Place.mapper = mapper(Place, place)
Transition.mapper = mapper(Transition, transition, properties = dict(
inputs = relation(Place.mapper, place_output, lazy=True, backref='inputs'),
outputs = relation(Place.mapper, place_input, lazy=True, backref='outputs'),
)
)
- #Place.mapper.add_property('inputs', relation(Transition.mapper, place_output, lazy=True, attributeext=attr.ListBackrefExtension('inputs')))
- #Place.mapper.add_property('outputs', relation(Transition.mapper, place_input, lazy=True, attributeext=attr.ListBackrefExtension('outputs')))
Place.eagermapper = Place.mapper.options(
eagerload('inputs', selectalias='ip_alias'),
enrolTbl.drop()
studentTbl.drop()
courseTbl.drop()
+ #testbase.db.tables.clear()
def setUp(self):
objectstore.clear()
c2a1.drop()
a.drop()
c.drop()
+ #testbase.db.tables.clear()
def testbasic(self):
class C(object):pass
class RelationTest(testbase.PersistTest):
- """this is essentially an extension of the "dependency.py" topological sort test. this exposes
- a particular issue that doesnt always occur with the straight dependency tests, due to the nature
- of the sort being different based on random conditions"""
+ """this is essentially an extension of the "dependency.py" topological sort test.
+ in this test, a table is dependent on two other tables that are otherwise unrelated to each other.
+ the dependency sort must insure that this childmost table is below both parent tables in the outcome
+ (a bug existed where this was not always the case).
+ while the straight topological sort tests should expose this, since the sorting can be different due
+ to subtle differences in program execution, this test case was exposing the bug whereas the simpler tests
+ were not."""
def setUpAll(self):
- testbase.db.tables.clear()
global tbl_a
global tbl_b
global tbl_c
tbl_c.drop()
tbl_b.drop()
tbl_a.drop()
+
+ def tearDownAll(self):
+ testbase.db.tables.clear()
def testDeleteRootTable(self):
session = objectstore.get_session()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
