Commits

Mike Bayer committed 2cc236e

- The "foreign_keys" argument of relation() will now propagate
automatically to the backref in the same way that
primaryjoin and secondaryjoin do. For the extremely
rare use case where the backref of a relation() has
intentionally different "foreign_keys" configured, both sides
now need to be configured explicity (if they do in fact require
this setting, see the next note...).

- ...the only known (and really, really rare) use case where a
different foreign_keys setting was used on the forwards/backwards
side, a composite foreign key that partially points to its own
columns, has been enhanced such that the fk->itself aspect of the
relation won't be used to determine relation direction.

Comments (0)

Files changed (3)

     - unit tests have been migrated from unittest to nose.
       See README.unittests for information on how to run
       the tests.  [ticket:970]
+      
 - orm
     - Fixed bug introduced in 0.5.4 whereby Composite types
       fail when default-holding columns are flushed.
       returned in terms of the base class rather than the subclass - 
       so applications which relied on this erroneous result need to be 
       adjusted. [ticket:1431]
+
+    - The "foreign_keys" argument of relation() will now propagate
+      automatically to the backref in the same way that
+      primaryjoin and secondaryjoin do.   For the extremely 
+      rare use case where the backref of a relation() has 
+      intentionally different "foreign_keys" configured, both sides
+      now need to be configured explicity (if they do in fact require
+      this setting, see the next note...).  
       
+    - ...the only known (and really, really rare) use case where a 
+      different foreign_keys setting was used on the forwards/backwards 
+      side, a composite foreign key that partially points to its own 
+      columns, has been enhanced such that the fk->itself aspect of the 
+      relation won't be used to determine relation direction.
+     
+    
 - sql
     - Removed an obscure feature of execute() (including connection,
       engine, Session) whereby a bindparam() construct can be sent as 

lib/sqlalchemy/orm/properties.py

                 elif l in self._foreign_keys:
                     self.synchronize_pairs.append((r, l))
         else:
-            eq_pairs = criterion_as_pairs(self.primaryjoin, consider_as_foreign_keys=self._foreign_keys, any_operator=self.viewonly)
-            eq_pairs = [(l, r) for l, r in eq_pairs if (self._col_is_part_of_mappings(l) and self._col_is_part_of_mappings(r)) or self.viewonly and r in self._foreign_keys]
+            eq_pairs = criterion_as_pairs(
+                            self.primaryjoin, 
+                            consider_as_foreign_keys=self._foreign_keys, 
+                            any_operator=self.viewonly
+                        )
+            eq_pairs = [
+                            (l, r) for l, r in eq_pairs if 
+                            (self._col_is_part_of_mappings(l) and 
+                                self._col_is_part_of_mappings(r)) 
+                                or self.viewonly and r in self._foreign_keys
+                        ]
 
             if not eq_pairs:
                 if not self.viewonly and criterion_as_pairs(self.primaryjoin, consider_as_foreign_keys=self._foreign_keys, any_operator=True):
         elif self._refers_to_parent_table():
             # self referential defaults to ONETOMANY unless the "remote" side is present
             # and does not reference any foreign key columns
+
             if self.local_remote_pairs:
                 remote = [r for l, r in self.local_remote_pairs]
             elif self.remote_side:
             else:
                 remote = None
 
-            if not remote or self._foreign_keys.intersection(remote):
+            if not remote or self._foreign_keys.\
+                                    difference(l for l, r in self.synchronize_pairs).\
+                                    intersection(remote):
                 self.direction = ONETOMANY
             else:
                 self.direction = MANYTOONE
                     raise sa_exc.InvalidRequestError(
                         "Can't assign 'secondaryjoin' on a backref against "
                         "a non-secondary relation.")
-
+            
+            foreign_keys = self.kwargs.pop('foreign_keys', prop._foreign_keys)
+            
             parent = prop.parent.primary_mapper()
             self.kwargs.setdefault('viewonly', prop.viewonly)
             self.kwargs.setdefault('post_update', prop.post_update)
 
-            relation = RelationProperty(parent, prop.secondary, pj, sj,
+            relation = RelationProperty(parent, prop.secondary, pj, sj, foreign_keys=foreign_keys,
                                       backref=BackRef(prop.key, _prop=prop),
                                       **self.kwargs)
 

test/orm/test_relationships.py

 from sqlalchemy import Integer, String, ForeignKey, MetaData, and_
 from sqlalchemy.test.schema import Table
 from sqlalchemy.test.schema import Column
-from sqlalchemy.orm import mapper, relation, backref, create_session, compile_mappers, clear_mappers
+from sqlalchemy.orm import mapper, relation, backref, create_session, compile_mappers, clear_mappers, sessionmaker
 from sqlalchemy.test.testing import eq_, startswith_
 from test.orm import _base, _fixtures
 
 
 
 class RelationTest2(_base.MappedTest):
-    """Tests a relationship on a column included in multiple foreign keys.
+    """The ultimate relation() test:
+    
+    company         employee
+    ----------      ----------
+    company_id <--- company_id ------+
+    name                ^            |
+                        +------------+
+                      
+                    emp_id <---------+
+                    name             |
+                    reports_to_id ---+
+    
+    employee joins to its sub-employees
+    both on reports_to_id, *and on company_id to itself*.
+    
+    As of 0.5.5 we are making a slight behavioral change,
+    such that the custom foreign_keys setting
+    on the o2m side has to be explicitly 
+    unset on the backref m2o side - this to suit
+    the vast majority of use cases where the backref()
+    is to receive the same foreign_keys argument 
+    as the forwards reference.   But we also
+    have smartened the remote_side logic such that 
+    you don't even need the custom fks setting.
+    
+    """
 
-    This test tests a relationship on a column that is included in multiple
-    foreign keys, as well as a self-referential relationship on a composite
-    key where one column in the foreign key is 'joined to itself'.
-
-    """
     @classmethod
     def define_tables(cls, metadata):
         Table('company_t', metadata,
                   ['company_id', 'reports_to_id'],
                   ['employee_t.company_id', 'employee_t.emp_id']))
 
-    @testing.resolve_artifact_names
-    def test_explicit(self):
-        """test with mappers that have fairly explicit join conditions"""
-
+    @classmethod
+    def setup_classes(cls):
         class Company(_base.Entity):
             pass
 
                 self.company = company
                 self.emp_id = emp_id
                 self.reports_to = reports_to
-
+        
+    @testing.resolve_artifact_names
+    def test_explicit(self):
         mapper(Company, company_t)
         mapper(Employee, employee_t, properties= {
             'company':relation(Company, primaryjoin=employee_t.c.company_id==company_t.c.company_id, backref='employees'),
                 ),
                 remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
                 foreign_keys=[employee_t.c.reports_to_id],
-                backref='employees')
+                backref=backref('employees', foreign_keys=None))
         })
 
-        sess = create_session()
-        c1 = Company()
-        c2 = Company()
-
-        e1 = Employee(u'emp1', c1, 1)
-        e2 = Employee(u'emp2', c1, 2, e1)
-        e3 = Employee(u'emp3', c1, 3, e1)
-        e4 = Employee(u'emp4', c1, 4, e3)
-        e5 = Employee(u'emp5', c2, 1)
-        e6 = Employee(u'emp6', c2, 2, e5)
-        e7 = Employee(u'emp7', c2, 3, e5)
-
-        sess.add_all((c1, c2))
-        sess.flush()
-        sess.expunge_all()
-
-        test_c1 = sess.query(Company).get(c1.company_id)
-        test_e1 = sess.query(Employee).get([c1.company_id, e1.emp_id])
-        assert test_e1.name == 'emp1', test_e1.name
-        test_e5 = sess.query(Employee).get([c2.company_id, e5.emp_id])
-        assert test_e5.name == 'emp5', test_e5.name
-        assert [x.name for x in test_e1.employees] == ['emp2', 'emp3']
-        eq_(sess.query(Employee).get([c1.company_id, 3]).reports_to.name, 'emp1')
-        eq_(sess.query(Employee).get([c2.company_id, 3]).reports_to.name, 'emp5')
+        self._test()
 
     @testing.resolve_artifact_names
     def test_implicit(self):
-        """test with mappers that have the most minimal arguments"""
-        class Company(_base.Entity):
-            pass
-        class Employee(_base.Entity):
-            def __init__(self, name, company, emp_id, reports_to=None):
-                self.name = name
-                self.company = company
-                self.emp_id = emp_id
-                self.reports_to = reports_to
-
         mapper(Company, company_t)
         mapper(Employee, employee_t, properties= {
             'company':relation(Company, backref='employees'),
             'reports_to':relation(Employee,
                 remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
                 foreign_keys=[employee_t.c.reports_to_id],
-                backref='employees')
+                backref=backref('employees', foreign_keys=None)
+                )
         })
 
+        self._test()
+
+    @testing.resolve_artifact_names
+    def test_very_implicit(self):
+        mapper(Company, company_t)
+        mapper(Employee, employee_t, properties= {
+            'company':relation(Company, backref='employees'),
+            'reports_to':relation(Employee,
+                remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
+                backref='employees'
+                )
+        })
+
+        self._test()
+    
+    @testing.resolve_artifact_names
+    def test_very_explicit(self):
+        mapper(Company, company_t)
+        mapper(Employee, employee_t, properties= {
+            'company':relation(Company, backref='employees'),
+            'reports_to':relation(Employee,
+                _local_remote_pairs = [
+                        (employee_t.c.reports_to_id, employee_t.c.emp_id),
+                        (employee_t.c.company_id, employee_t.c.company_id)
+                ],
+                foreign_keys=[employee_t.c.reports_to_id],
+                backref=backref('employees', foreign_keys=None)
+                )
+        })
+
+        self._test()
+        
+    @testing.resolve_artifact_names
+    def _test(self):
         sess = create_session()
         c1 = Company()
         c2 = Company()
             [TagInstance(data='iplc_case'), TagInstance(data='not_iplc_case')]
         )
 
+class BackrefPropagatesForwardsArgs(_base.MappedTest):
+    
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('users', metadata, 
+            Column('id', Integer, primary_key=True),
+            Column('name', String(50))
+        )
+        Table('addresses', metadata, 
+            Column('id', Integer, primary_key=True),
+            Column('user_id', Integer),
+            Column('email', String(50))
+        )
+    
+    @classmethod
+    def setup_classes(cls):
+        class User(_base.ComparableEntity):
+            pass
+        class Address(_base.ComparableEntity):
+            pass
+    
+    @testing.resolve_artifact_names
+    def test_backref(self):
+        
+        mapper(User, users, properties={
+            'addresses':relation(Address, 
+                        primaryjoin=addresses.c.user_id==users.c.id, 
+                        foreign_keys=addresses.c.user_id,
+                        backref='user')
+        })
+        mapper(Address, addresses)
+        
+        sess = sessionmaker()()
+        u1 = User(name='u1', addresses=[Address(email='a1')])
+        sess.add(u1)
+        sess.commit()
+        eq_(sess.query(Address).all(), [
+            Address(email='a1', user=User(name='u1'))
+        ])
+    
 class AmbiguousJoinInterpretedAsSelfRef(_base.MappedTest):
     """test ambiguous joins due to FKs on both sides treated as self-referential.