return len(self.dependencies) == 0 or (len(self.dependencies)==1 and list(self.dependencies)[0] == self.mapper.class_key)
def compiled(self, classkey):
self.dependencies.remove(classkey)
-
+ def __str__(self):
+ return "CompileTrigger on mapper " + str(self.mapper)
+
class Mapper(object):
"""Persists object instances to and from schema.Table objects via the sql package.
Instances of this class should be constructed through this package's mapper() or
triggerset = _compile_triggers.pop(self.class_key, None)
if triggerset is not None:
for rec in triggerset:
+ print "looking at:", rec
rec.compiled(self.class_key)
if rec.can_compile():
rec.mapper._do_compile()
except (KeyError, AttributeError):
raise exceptions.InvalidRequestError("Class '%s' entity name '%s' has no mapper associated with it" % (class_.__name__, entity_name))
if compile:
- return mapper._do_compile()
+ return mapper.compile()
else:
return mapper
except exceptions.ArgumentError, e:
assert str(e).index("Backrefs do not match") > -1
+ def testthree(self):
+ metadata = BoundMetaData(testbase.db)
+ node_table = Table("node", metadata,
+ Column('node_id', Integer, primary_key=True),
+ Column('name_index', Integer, nullable=True),
+ )
+ node_name_table = Table("node_name", metadata,
+ Column('node_name_id', Integer, primary_key=True),
+ Column('node_id', Integer, ForeignKey('node.node_id')),
+ Column('host_id', Integer, ForeignKey('host.host_id')),
+ Column('name', String(64), nullable=False),
+ )
+ host_table = Table("host", metadata,
+ Column('host_id', Integer, primary_key=True),
+ Column('hostname', String(64), nullable=False,
+ unique=True),
+ )
+ metadata.create_all()
+ try:
+ node_table.insert().execute(node_id=1, node_index=5)
+ class Node(object):pass
+ class NodeName(object):pass
+ class Host(object):pass
+
+ node_mapper = mapper(Node, node_table)
+ host_mapper = mapper(Host, host_table)
+ node_name_mapper = mapper(NodeName, node_name_table,
+ properties = {
+ 'node' : relation(Node, backref=backref('names')),
+ 'host' : relation(Host),
+ }
+ )
+ sess = create_session()
+ assert sess.query(Node).get(1).names == []
+ finally:
+ metadata.drop_all()
+
if __name__ == '__main__':
testbase.main()