From: Mike Bayer Date: Fri, 7 Mar 2008 03:16:46 +0000 (+0000) Subject: - moved property._is_self_referential() to be more generalized; returns True for... X-Git-Tag: rel_0_4_4~22 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=f621df6ce6f3d60ec5badf789f98f62d1d0c206c;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git - moved property._is_self_referential() to be more generalized; returns True for any mapper.isa() relationship between parent and child, and indicates that aliasing should be used for any join/correlation across the relation. allows joins/any()/has() to work with inherited mappers referencing the parent etc. - the original _is_self_referential() is now _refers_to_parent_table() and is only used during "direction" calculation to indicate the relation is from a single table to itself --- diff --git a/CHANGES b/CHANGES index 1295881976..0fe58815fb 100644 --- a/CHANGES +++ b/CHANGES @@ -31,7 +31,9 @@ CHANGES - any(), has(), contains(), ~contains(), attribute level == and != now work properly with self-referential relations - the clause inside the EXISTS is aliased on the "remote" - side to distinguish it from the parent table. + side to distinguish it from the parent table. This + applies to single table self-referential as well as + inheritance-based self-referential. - repaired behavior of == and != operators at the relation() level when compared against NULL for one-to-one diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py index ba5ef7d364..8e17157da7 100644 --- a/lib/sqlalchemy/orm/properties.py +++ b/lib/sqlalchemy/orm/properties.py @@ -546,7 +546,7 @@ class PropertyLoader(StrategizedProperty): self.secondary.c.contains_column(column) is not None def _determine_fks(self): - if self._legacy_foreignkey and not self._is_self_referential(): + if self._legacy_foreignkey and not self._refers_to_parent_table(): self.foreign_keys = self._legacy_foreignkey if self.foreign_keys: @@ -601,7 +601,7 @@ class PropertyLoader(StrategizedProperty): if self.secondaryjoin is not None: self.direction = sync.MANYTOMANY - elif self._is_self_referential(): + elif self._refers_to_parent_table(): # for a self referential mapper, if the "foreignkey" is a single or composite primary key, # then we are "many to one", since the remote site of the relationship identifies a singular entity. # otherwise we are "one to many". @@ -728,9 +728,12 @@ class PropertyLoader(StrategizedProperty): super(PropertyLoader, self).do_init() - def _is_self_referential(self): + def _refers_to_parent_table(self): return self.parent.mapped_table is self.target or self.parent.select_table is self.target - + + def _is_self_referential(self): + return self.mapper.isa(self.parent) + def primary_join_against(self, mapper, selectable=None, toselectable=None): return self.__cached_join_against(mapper, selectable, toselectable, True, False) diff --git a/test/orm/generative.py b/test/orm/generative.py index c917b3e6e1..aced8f626f 100644 --- a/test/orm/generative.py +++ b/test/orm/generative.py @@ -249,17 +249,15 @@ class SelfRefTest(ORMTest): class T(object):pass mapper(T, t1, properties={'children':relation(T)}) sess = create_session(bind=testing.db) - try: - sess.query(T).join('children').select_by(id=7) - assert False - except exceptions.InvalidRequestError, e: - assert str(e) == "Self-referential query on 'T.children (T)' property requires aliased=True argument.", str(e) + def go(): + sess.query(T).join('children') + self.assertRaisesMessage(exceptions.InvalidRequestError, + "Self-referential query on 'T\.children \(T\)' property requires aliased=True argument.", go) - try: + def go(): sess.query(T).join(['children']).select_by(id=7) - assert False - except exceptions.InvalidRequestError, e: - assert str(e) == "Self-referential query on 'T.children (T)' property requires aliased=True argument.", str(e) + self.assertRaisesMessage(exceptions.InvalidRequestError, + "Self-referential query on 'T\.children \(T\)' property requires aliased=True argument.", go) diff --git a/test/orm/inheritance/query.py b/test/orm/inheritance/query.py index 7d7b8b9d91..777913b08e 100644 --- a/test/orm/inheritance/query.py +++ b/test/orm/inheritance/query.py @@ -7,6 +7,7 @@ import testenv; testenv.configure_for_tests() import sets from sqlalchemy import * from sqlalchemy.orm import * +from sqlalchemy import exceptions from testlib import * from testlib import fixtures @@ -375,5 +376,57 @@ for select_type in ('', 'Unions', 'AliasedJoins', 'Joins'): del testclass +class SelfReferentialTest(ORMTest): + keep_mappers = True + + def define_tables(self, metadata): + global people, engineers + 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))) + + engineers = Table('engineers', metadata, + Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), + Column('primary_language', String(50)), + Column('reports_to_id', Integer, ForeignKey('people.person_id')) + ) + + mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person') + mapper(Engineer, engineers, inherits=Person, + inherit_condition=engineers.c.person_id==people.c.person_id, + polymorphic_identity='engineer', properties={ + 'reports_to':relation(Person, primaryjoin=people.c.person_id==engineers.c.reports_to_id) + }) + + def test_has(self): + + p1 = Person(name='dogbert') + e1 = Engineer(name='dilbert', primary_language='java', reports_to=p1) + sess = create_session() + sess.save(p1) + sess.save(e1) + sess.flush() + sess.clear() + + self.assertEquals(sess.query(Engineer).filter(Engineer.reports_to.has(Person.name=='dogbert')).first(), Engineer(name='dilbert')) + + def test_join(self): + p1 = Person(name='dogbert') + e1 = Engineer(name='dilbert', primary_language='java', reports_to=p1) + sess = create_session() + sess.save(p1) + sess.save(e1) + sess.flush() + sess.clear() + + self.assertEquals(sess.query(Engineer).join('reports_to', aliased=True).filter(Person.name=='dogbert').first(), Engineer(name='dilbert')) + + def test_noalias_raises(self): + sess = create_session() + def go(): + sess.query(Engineer).join('reports_to') + self.assertRaises(exceptions.InvalidRequestError, go) + if __name__ == "__main__": testenv.main()