From a31979937223c68f395ee0704593c9150f10e86b Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Sat, 20 Jan 2007 12:31:27 +0000 Subject: [PATCH] - tightened down conditions used to locate "relation direction", associating the "foreignkey" of the relationship with the "primaryjoin". the column match now must be exact, not just "corresponding". this enables self-referential relationships on a polymorphic mapper. - a little bit of improvement to the concept of a "concrete" inheritance mapping, though that concept is not well fleshed out yet (added test case to support concrete mappers on top of a polymorphic base). --- CHANGES | 5 +- lib/sqlalchemy/orm/mapper.py | 4 +- lib/sqlalchemy/orm/properties.py | 4 +- lib/sqlalchemy/orm/strategies.py | 2 +- lib/sqlalchemy/sql.py | 10 ++- test/orm/alltests.py | 2 + test/orm/inheritance4.py | 71 ++++++++++++++++ test/orm/inheritance5.py | 135 +++++++++++++++++++++++++++++++ test/orm/relationships.py | 69 ---------------- 9 files changed, 227 insertions(+), 75 deletions(-) create mode 100644 test/orm/inheritance4.py create mode 100644 test/orm/inheritance5.py diff --git a/CHANGES b/CHANGES index 28a3c1270b..67b0e4f181 100644 --- a/CHANGES +++ b/CHANGES @@ -54,7 +54,10 @@ actual primary key in "someinstance". - some deeper error checking when compiling relations, to detect an ambiguous "primaryjoin" in the case that both sides of the relationship have foreign key references in the primary - join condition + join condition. also tightened down conditions used to locate "relation direction", associating + the "foreignkey" of the relationship with the "primaryjoin" + - a little bit of improvement to the concept of a "concrete" inheritance mapping, though that concept + is not well fleshed out yet (added test case to support concrete mappers on top of a polymorphic base). - fix to "proxy=True" behavior on synonym() - fixed bug where delete-orphan basically didn't work with many-to-many relationships [ticket:427], backref presence generally hid the symptom diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py index 872f3d1212..8efd70967e 100644 --- a/lib/sqlalchemy/orm/mapper.py +++ b/lib/sqlalchemy/orm/mapper.py @@ -462,7 +462,7 @@ class Mapper(object): if self.inherits is not None: for key, prop in self.inherits.__props.iteritems(): - if not self.__props.has_key(key): + if not self.__props.has_key(key) and (not self.concrete or not isinstance(prop, ColumnProperty)): prop.adapt_to_inherited(key, self) # load properties from the main table object, @@ -516,6 +516,8 @@ class Mapper(object): the columns of select_table should encompass all the columns of the mapped_table either directly or through proxying relationships.""" if self.select_table is not self.mapped_table: + if self.polymorphic_identity is None: + raise exceptions.ArgumentError("Could not locate a polymorphic_identity field for mapper '%s'. This field is required for polymorphic mappers" % str(self)) props = {} if self.properties is not None: for key, prop in self.properties.iteritems(): diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py index 452af04cc9..db43a8e27a 100644 --- a/lib/sqlalchemy/orm/properties.py +++ b/lib/sqlalchemy/orm/properties.py @@ -274,8 +274,8 @@ class PropertyLoader(StrategizedProperty): else: return sync.MANYTOONE else: - onetomany = len([c for c in self.foreignkey if self.mapper.unjoined_table.corresponding_column(c, False) is not None]) - manytoone = len([c for c in self.foreignkey if self.parent.unjoined_table.corresponding_column(c, False) is not None]) + onetomany = len([c for c in self.foreignkey if self.mapper.unjoined_table.corresponding_column(c, False, require_exact=True) is not None]) + manytoone = len([c for c in self.foreignkey if self.parent.unjoined_table.corresponding_column(c, False, require_exact=True) is not None]) if not onetomany and not manytoone: raise exceptions.ArgumentError("Cant determine relation direction for '%s' on mapper '%s' with primary join '%s' - foreign key columns are not present in neither the parent nor the child's mapped tables" %(self.key, str(self.parent), str(self.primaryjoin))) elif onetomany and manytoone: diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py index 54590f4adf..88a22bbcdd 100644 --- a/lib/sqlalchemy/orm/strategies.py +++ b/lib/sqlalchemy/orm/strategies.py @@ -248,7 +248,7 @@ class LazyLoader(AbstractRelationLoader): binds = {} reverse = {} def column_in_table(table, column): - return table.corresponding_column(column, raiseerr=False, keys_ok=False) is not None + return table.corresponding_column(column, raiseerr=False, keys_ok=False, require_exact=True) is not None if remote_side is None or len(remote_side) == 0: remote_side = foreignkey diff --git a/lib/sqlalchemy/sql.py b/lib/sqlalchemy/sql.py index f944b468d6..5b960546e6 100644 --- a/lib/sqlalchemy/sql.py +++ b/lib/sqlalchemy/sql.py @@ -728,9 +728,17 @@ class FromClause(Selectable): if not hasattr(self, '_oid_column'): self._oid_column = self._locate_oid_column() return self._oid_column - def corresponding_column(self, column, raiseerr=True, keys_ok=False): + def corresponding_column(self, column, raiseerr=True, keys_ok=False, require_exact=False): """given a ColumnElement, return the ColumnElement object from this Selectable which corresponds to that original Column via a proxy relationship.""" + if require_exact: + if self.columns.get(column.key) is column: + return column + else: + if not raiseerr: + return None + else: + raise exceptions.InvalidRequestError("Column instance '%s' is not directly present in table '%s'" % (str(column), str(column.table))) for c in column.orig_set: try: return self.original_columns[c] diff --git a/test/orm/alltests.py b/test/orm/alltests.py index 3fbdbadfc8..ffcdd3d218 100644 --- a/test/orm/alltests.py +++ b/test/orm/alltests.py @@ -28,6 +28,8 @@ def suite(): 'orm.inheritance', 'orm.inheritance2', 'orm.inheritance3', + 'orm.inheritance4', + 'orm.inheritance5', 'orm.single', 'orm.polymorph' ) diff --git a/test/orm/inheritance4.py b/test/orm/inheritance4.py new file mode 100644 index 0000000000..7a97d06c14 --- /dev/null +++ b/test/orm/inheritance4.py @@ -0,0 +1,71 @@ +from sqlalchemy import * +import testbase + +class ConcreteTest1(testbase.AssertMixin): + def setUpAll(self): + global managers_table, engineers_table, metadata + metadata = BoundMetaData(testbase.db) + managers_table = Table('managers', metadata, + Column('employee_id', Integer, primary_key=True), + Column('name', String(50)), + Column('manager_data', String(50)), + ) + + engineers_table = Table('engineers', metadata, + Column('employee_id', Integer, primary_key=True), + Column('name', String(50)), + Column('engineer_info', String(50)), + ) + + metadata.create_all() + def tearDownAll(self): + metadata.drop_all() + + def testbasic(self): + class Employee(object): + def __init__(self, name): + self.name = name + def __repr__(self): + return self.__class__.__name__ + " " + self.name + + class Manager(Employee): + def __init__(self, name, manager_data): + self.name = name + self.manager_data = manager_data + def __repr__(self): + return self.__class__.__name__ + " " + self.name + " " + self.manager_data + + class Engineer(Employee): + def __init__(self, name, engineer_info): + self.name = name + self.engineer_info = engineer_info + def __repr__(self): + return self.__class__.__name__ + " " + self.name + " " + self.engineer_info + + pjoin = polymorphic_union({ + 'manager':managers_table, + 'engineer':engineers_table + }, 'type', 'pjoin') + + employee_mapper = mapper(Employee, pjoin, polymorphic_on=pjoin.c.type) + manager_mapper = mapper(Manager, managers_table, inherits=employee_mapper, concrete=True, polymorphic_identity='manager') + engineer_mapper = mapper(Engineer, engineers_table, inherits=employee_mapper, concrete=True, polymorphic_identity='engineer') + + session = create_session() + session.save(Manager('Tom', 'knows how to manage things')) + session.save(Engineer('Kurt', 'knows how to hack')) + session.flush() + session.clear() + + assert set([repr(x) for x in session.query(Employee).select()]) == set(["Engineer Kurt knows how to hack", "Manager Tom knows how to manage things"]) + assert set([repr(x) for x in session.query(Manager).select()]) == set(["Manager Tom knows how to manage things"]) + assert set([repr(x) for x in session.query(Engineer).select()]) == set(["Engineer Kurt knows how to hack"]) + + def testwithrelation(self): + pass + + # TODO: test a self-referential relationship on a concrete polymorphic mapping + + +if __name__ == '__main__': + testbase.main() \ No newline at end of file diff --git a/test/orm/inheritance5.py b/test/orm/inheritance5.py new file mode 100644 index 0000000000..e54bcb8c83 --- /dev/null +++ b/test/orm/inheritance5.py @@ -0,0 +1,135 @@ +from sqlalchemy import * +import testbase + +class AttrSettable(object): + def __init__(self, **kwargs): + [setattr(self, k, v) for k, v in kwargs.iteritems()] + def __repr__(self): + return self.__class__.__name__ + ' ' + ','.join(["%s=%s" % (k,v) for k, v in self.__dict__.iteritems() if k[0] != '_']) + + +class RelationTest1(testbase.PersistTest): + """test self-referential relationships on polymorphic mappers""" + def setUpAll(self): + global people, managers, metadata + metadata = BoundMetaData(testbase.db) + + people = Table('people', metadata, + Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True), + Column('manager_id', Integer, ForeignKey('managers.person_id', use_alter=True, name="mpid_fq")), + Column('name', String(50)), + Column('type', String(30))) + + managers = Table('managers', metadata, + Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), + Column('status', String(30)), + Column('manager_name', String(50)) + ) + + metadata.create_all() + + def tearDownAll(self): + metadata.drop_all() + + def tearDown(self): + clear_mappers() + people.update().execute(manager_id=None) + for t in metadata.table_iterator(reverse=True): + t.delete().execute() + + def testbasic(self): + class Person(AttrSettable): + pass + class Manager(Person): + pass + + mapper(Person, people, properties={ + 'manager':relation(Manager, primaryjoin=people.c.manager_id==managers.c.person_id, uselist=False) + }) + mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id) + try: + compile_mappers() + except exceptions.ArgumentError, ar: + assert str(ar) == "Cant determine relation direction for 'manager' on mapper 'Mapper|Person|people' with primary join 'people.manager_id = managers.person_id' - foreign key columns are present in both the parent and the child's mapped tables. Specify 'foreignkey' argument." + + clear_mappers() + + mapper(Person, people, properties={ + 'manager':relation(Manager, primaryjoin=people.c.manager_id==managers.c.person_id, foreignkey=people.c.manager_id, uselist=False, post_update=True) + }) + mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id) + + session = create_session() + p = Person(name='some person') + m = Manager(name='some manager') + p.manager = m + session.save(p) + session.flush() + session.clear() + + p = session.query(Person).get(p.person_id) + m = session.query(Manager).get(m.person_id) + print p, m, p.manager + assert p.manager is m + +class RelationTest2(testbase.AssertMixin): + """test self-referential relationships on polymorphic mappers""" + def setUpAll(self): + global people, managers, metadata + metadata = BoundMetaData(testbase.db) + + people = Table('people', metadata, + Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True), + Column('name', String(50)), + Column('type', String(30))) + + managers = Table('managers', metadata, + Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), + Column('manager_id', Integer, ForeignKey('people.person_id')), + Column('status', String(30)), + ) + + metadata.create_all() + + def tearDownAll(self): + metadata.drop_all() + + def tearDown(self): + clear_mappers() + for t in metadata.table_iterator(reverse=True): + t.delete().execute() + + def testbasic(self): + class Person(AttrSettable): + pass + class Manager(Person): + pass + + poly_union = polymorphic_union({ + 'person':people.select(people.c.type=='person'), + 'manager':managers.join(people, people.c.person_id==managers.c.person_id) + }, None) + + mapper(Person, people, select_table=poly_union, polymorphic_identity='person', polymorphic_on=people.c.type) + mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id, polymorphic_identity='manager', + properties={ + 'colleague':relation(Person, primaryjoin=managers.c.manager_id==people.c.person_id, uselist=False) + }) + + sess = create_session() + p = Person(name='person1') + m = Manager(name='manager1') + m.colleague = p + sess.save(m) + sess.flush() + + sess.clear() + p = sess.query(Person).get(p.person_id) + m = sess.query(Manager).get(m.person_id) + print p + print m + assert m.colleague is p + +if __name__ == "__main__": + testbase.main() + \ No newline at end of file diff --git a/test/orm/relationships.py b/test/orm/relationships.py index ef8ee9e02b..250f5ab0d5 100644 --- a/test/orm/relationships.py +++ b/test/orm/relationships.py @@ -351,75 +351,6 @@ class RelationTest3(testbase.PersistTest): s.delete(j) s.flush() -class CyclicalRelationInheritTest(testbase.PersistTest): - """testing a mapper with an inheriting mapper, where the base mapper also has a relation to the inheriting mapper, - as well as using unexpected columns to create the join conditions which do not indicate the usual "cyclical" relationship - this represents. test that proper error messages are raised guiding the user to a reasonably working setup, and that - the ultimate setup works.""" - def setUpAll(self): - global people, managers, metadata - metadata = BoundMetaData(testbase.db) - - people = Table('people', metadata, - Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True), - Column('manager_id', Integer, ForeignKey('managers.person_id', use_alter=True, name="mpid_fq")), - Column('name', String(50)), - Column('type', String(30))) - - managers = Table('managers', metadata, - Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), - Column('status', String(30)), - Column('manager_name', String(50)) - ) - - metadata.create_all() - - def tearDownAll(self): - metadata.drop_all() - - def tearDown(self): - clear_mappers() - people.update().execute(manager_id=None) - for t in metadata.table_iterator(reverse=True): - t.delete().execute() - - def testbasic(self): - class Person(object): - pass - class Manager(Person): - pass - - mapper(Person, people, properties={ - 'manager':relation(Manager, primaryjoin=people.c.manager_id==managers.c.person_id, uselist=False) - }) - mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id) - try: - compile_mappers() - except exceptions.ArgumentError, ar: - assert str(ar) == "Cant determine relation direction for 'manager' on mapper 'Mapper|Person|people' with primary join 'people.manager_id = managers.person_id' - foreign key columns are present in both the parent and the child's mapped tables. Specify 'foreignkey' argument." - - clear_mappers() - - mapper(Person, people, properties={ - 'manager':relation(Manager, primaryjoin=people.c.manager_id==managers.c.person_id, foreignkey=people.c.manager_id, uselist=False, post_update=True) - }) - mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id) - - session = create_session() - p = Person() - p.name = 'some person' - m = Manager() - m.name = 'some manager' - p.manager = m - session.save(p) - session.flush() - session.clear() - - p = session.query(Person).get(p.person_id) - m = session.query(Manager).get(m.person_id) - print p, m, p.manager - assert p.manager is m - if __name__ == "__main__": -- 2.47.2