class TableFinder(sql.ClauseVisitor):
"""given a Clause, locates all the Tables within it into a list."""
- def __init__(self, table):
+ def __init__(self, table, check_columns=False):
self.tables = []
+ self.check_columns = check_columns
table.accept_visitor(self)
def visit_table(self, table):
self.tables.append(table)
return obj in self.tables
def __add__(self, obj):
return self.tables + obj
-
+ def visit_column(self, column):
+ if self.check_columns:
+ column.table.accept_visitor(self)
+
def hash_key(obj):
if obj is None:
return 'None'
self.secondaryjoin = sql.join(self.target, self.secondary).onclause
if self.primaryjoin is None:
self.primaryjoin = sql.join(parent.table, self.secondary).onclause
+ tf = mapper.TableFinder(self.secondaryjoin, check_columns=True)
+ tf2 = mapper.TableFinder(self.primaryjoin, check_columns=True)
+ for t in tf2:
+ if t is not self.secondary and t in tf:
+ raise ArgumentError("Ambiguous join conditions generated between '%s' and '%s' (primaryjoin='%s', secondaryjoin='%s'); please specify explicit primaryjoin and/or secondaryjoin arguments to property '%s'" % (parent.table.id, self.target.id, self.primaryjoin, self.secondaryjoin, self.key))
else:
if self.primaryjoin is None:
self.primaryjoin = sql.join(parent.table, self.target).onclause
table.drop()
testbase.db.tables.clear()
+ def tearDown(self):
+ for table in reversed(tables):
+ table.delete().execute()
+
def testbasic(self):
class Foo(object):
+ def __init__(self, data=None):
+ self.data = data
def __repr__(self):
return "Foo id %d, data %s" % (self.id, self.data)
Foo.mapper = mapper(Foo, foo)
- class Bar(object):
+ class Bar(Foo):
def __repr__(self):
return "Bar id %d, data %s" % (self.id, self.data)
Bar.mapper = mapper(Bar, bar, inherits=Foo.mapper, properties={
- 'foos' :relation(Foo.mapper, bar_foo, primaryjoin=bar.c.id==bar_foo.c.bar_id, secondaryjoin=bar_foo.c.foo_id==foo.c.id, lazy=False)
+ 'foos' :relation(Foo.mapper, bar_foo, primaryjoin=bar.c.id==bar_foo.c.bar_id, lazy=False)
+# 'foos' :relation(Foo.mapper, bar_foo, lazy=True)
})
- Bar.mapper.select()
+ b = Bar('bar #1')
+ b.foos.append(Foo("foo #1"))
+ b.foos.append(Foo("foo #2"))
+ objectstore.commit()
+ objectstore.clear()
+ l = Bar.mapper.select()
+ print l[0], l[0].foos
def testadvanced(self):
class Foo(object):
return "Blub id %d, data %s, bars %s, foos %s" % (self.id, self.data, repr([b for b in self.bars]), repr([f for f in self.foos]))
Blub.mapper = mapper(Blub, blub, inherits=Bar.mapper, properties={
- 'bars':relation(Bar.mapper, blub_bar, primaryjoin=blub.c.id==blub_bar.c.blub_id, secondaryjoin=blub_bar.c.bar_id==bar.c.id, lazy=False),
- 'foos':relation(Foo.mapper, blub_foo, primaryjoin=blub.c.id==blub_foo.c.blub_id, secondaryjoin=blub_foo.c.foo_id==foo.c.id, lazy=False),
+ 'bars':relation(Bar.mapper, blub_bar, primaryjoin=blub.c.id==blub_bar.c.blub_id, lazy=False),
+ 'foos':relation(Foo.mapper, blub_foo, primaryjoin=blub.c.id==blub_foo.c.blub_id, lazy=False),
})
useobjects = True
blub_foo.insert().execute(blub_id=1, foo_id=2)
l = Blub.mapper.select()
- for x in l:
- print x
-
+ self.echo(l)
self.assert_(repr(l[0]) == compare)
objectstore.clear()
x = Blub.mapper.get_by(id=blubid) #traceback 2
+ self.echo(x)
self.assert_(repr(x) == compare)
+
if __name__ == "__main__":