Ticket #9475: 9475.m2m_add_remove.r12009.diff

tests/modeltests/m2m_through/models.py

@@ -122 +122 @@
 
 ### Forward Descriptors Tests ###
 
-# Due to complications with adding via an intermediary model,
-# the add method is not provided.
->>> rock.members.add(bob)
-Traceback (most recent call last):
-...
-AttributeError: 'ManyRelatedManager' object has no attribute 'add'
-
-# Create is also disabled as it suffers from the same problems as add.
->>> rock.members.create(name='Anne')
-Traceback (most recent call last):
-...
-AttributeError: Cannot use create() on a ManyToManyField which specifies an intermediary model. Use m2m_through.Membership's Manager instead.
-
-# Remove has similar complications, and is not provided either.
->>> rock.members.remove(jim)
-Traceback (most recent call last):
-...
-AttributeError: 'ManyRelatedManager' object has no attribute 'remove'
-
-# Here we back up the list of all members of Rock.
->>> backup = list(rock.members.all())
-
-# ...and we verify that it has worked.
->>> backup
-[<Person: Jane>, <Person: Jim>]
-
-# The clear function should still work.
+# The clear function should work.
 >>> rock.members.clear()
 
 # Now there will be no members of Rock.
 >>> rock.members.all()
 []
 
-# Assignment should not work with models specifying a through model for many of
-# the same reasons as adding.
->>> rock.members = backup
-Traceback (most recent call last):
-...
-AttributeError: Cannot set values on a ManyToManyField which specifies an intermediary model.  Use m2m_through.Membership's Manager instead.
-
 # Let's re-save those instances that we've cleared.
 >>> m1.save()
 >>> m2.save()
@@ -173 +140 @@
 
 ### Reverse Descriptors Tests ###
 
-# Due to complications with adding via an intermediary model,
-# the add method is not provided.
->>> bob.group_set.add(rock)
-Traceback (most recent call last):
-...
-AttributeError: 'ManyRelatedManager' object has no attribute 'add'
-
-# Create is also disabled as it suffers from the same problems as add.
->>> bob.group_set.create(name='Funk')
-Traceback (most recent call last):
-...
-AttributeError: Cannot use create() on a ManyToManyField which specifies an intermediary model. Use m2m_through.Membership's Manager instead.
-
-# Remove has similar complications, and is not provided either.
->>> jim.group_set.remove(rock)
-Traceback (most recent call last):
-...
-AttributeError: 'ManyRelatedManager' object has no attribute 'remove'
-
-# Here we back up the list of all of Jim's groups.
->>> backup = list(jim.group_set.all())
->>> backup
-[<Group: Rock>, <Group: Roll>]
-
-# The clear function should still work.
+# The clear function should work.
 >>> jim.group_set.clear()
 
 # Now Jim will be in no groups.
 >>> jim.group_set.all()
 []
 
-# Assignment should not work with models specifying a through model for many of
-# the same reasons as adding.
->>> jim.group_set = backup
-Traceback (most recent call last):
-...
-AttributeError: Cannot set values on a ManyToManyField which specifies an intermediary model.  Use m2m_through.Membership's Manager instead.
-
 # Let's re-save those instances that we've cleared.
 >>> m1.save()
 >>> m4.save()

tests/modeltests/m2m_add_and_remove/__init__.py

@@ +1 @@
+
+

tests/modeltests/m2m_add_and_remove/models.py

@@ +1 @@
+from django.db import models
+from django.test import TestCase
+
+
+class M(models.Model):
+    default = models.ManyToManyField("R", related_name="default_m_set")
+    default_cannot_remove = models.ManyToManyField("R", can_remove=False, related_name="default_cannot_remove_m_set")
+    default_cannot_add = models.ManyToManyField("R", can_add=False, related_name="default_cannot_add_m_set")
+    through_default = models.ManyToManyField("R", through="ThroughDefault", related_name="through_default_m_set")
+    through_auto = models.ManyToManyField("R", through="ThroughAuto", related_name="through_auto_m_set")
+    through_ut = models.ManyToManyField("R", through="ThroughUT", related_name="through_ut_m_set")
+    through_can_add = models.ManyToManyField("R", can_add=True, through="ThroughCanAdd", related_name="through_can_add_m_set")
+    through_can_remove = models.ManyToManyField("R", can_remove=True, through="ThroughCanRemove", related_name="through_can_remove_m_set")
+
+class R(models.Model):
+    name = models.CharField(max_length=30)
+
+class Through(models.Model):
+    m = models.ForeignKey(M, related_name="%(class)s_set")
+    r = models.ForeignKey(R, related_name="%(class)s_set")
+
+    class Meta:
+        abstract = True
+        
+class ThroughDefault(Through):
+    extra = models.CharField(max_length=10)
+
+class ThroughAuto(Through):
+    ctime = models.DateTimeField(auto_now_add=True)
+    mtime = models.DateTimeField(auto_now=True)
+    default = models.IntegerField(default=42)
+    null = models.DateTimeField(null=True)
+    
+class ThroughUT(Through):
+    extra = models.CharField(max_length=10)
+
+    class Meta:
+        unique_together = ('m', 'r')
+        
+class ThroughCanRemove(Through):
+    extra = models.CharField(max_length=10)
+
+class ThroughCanAdd(Through):
+    extra = models.CharField(max_length=10)
+    
+    def save(self, **kwargs):
+        self.extra = "foo"
+        return super(ThroughCanAdd, self).save(**kwargs)
+
+class M2mAddRemoveTests(TestCase):
+    def assert_cannot_remove(self, name):
+        m = M.objects.create()
+        r = R.objects.create()
+        manager = getattr(m, name)
+        reverse_manager = getattr(r, "%s_m_set" % name)
+        self.assertRaises(AttributeError, getattr, manager, 'remove')
+        self.assertRaises(AttributeError, getattr, reverse_manager, 'remove')
+        
+    def assert_cannot_add(self, name):
+        reverse_name = "%s_m_set" % name
+        m = M.objects.create()
+        r = R.objects.create()
+        manager = getattr(m, name)
+        reverse_manager = getattr(r, reverse_name)
+        self.assertRaises(AttributeError, getattr, manager, 'add')
+        self.assertRaises(AttributeError, getattr, reverse_manager, 'add')
+        self.assertRaises(AttributeError, manager.create)
+        self.assertRaises(AttributeError, reverse_manager.create)
+        def assign():
+            setattr(m, name, [])
+        self.assertRaises(AttributeError, assign)
+        def assign_reverse():
+            setattr(r, reverse_name, [])
+        self.assertRaises(AttributeError, assign_reverse)
+        
+    def assert_can_add(self, name):
+        reverse_name = "%s_m_set" % name
+        m = M.objects.create()
+        r = R.objects.create()
+        manager = getattr(m, name)
+        reverse_manager = getattr(r, reverse_name)
+        
+        manager.add(r)
+        self.failUnlessEqual(list(manager.all()), [r])
+        self.failUnlessEqual(list(reverse_manager.all()), [m])
+        manager.add(r)
+        self.failUnlessEqual(list(manager.all()), [r])
+        self.failUnlessEqual(list(reverse_manager.all()), [m])
+        manager.clear()
+        
+        reverse_manager.add(m)
+        self.failUnlessEqual(list(manager.all()), [r])
+        self.failUnlessEqual(list(reverse_manager.all()), [m])
+        reverse_manager.add(m)
+        self.failUnlessEqual(list(manager.all()), [r])
+        self.failUnlessEqual(list(reverse_manager.all()), [m])
+        reverse_manager.clear()
+        
+        r2 = manager.create()
+        reverse_manager2 = getattr(r2, reverse_name)
+        self.failUnlessEqual(list(manager.all()), [r2])
+        self.failUnlessEqual(list(reverse_manager2.all()), [m])
+        manager.clear()
+        
+        m2 = reverse_manager.create()
+        manager2 = getattr(m2, name)
+        self.failUnlessEqual(list(manager2.all()), [r])
+        self.failUnlessEqual(list(reverse_manager.all()), [m2])
+        reverse_manager.clear()
+        
+        setattr(m, name, [r])
+        self.failUnlessEqual(list(manager.all()), [r])
+        manager.clear()
+        
+        setattr(r, reverse_name, [m])
+        self.failUnlessEqual(list(reverse_manager.all()), [m])
+        reverse_manager.clear()
+        
+    def assert_can_remove(self, name, extra):
+        through = M._meta.get_field(name).rel.through
+        m = M.objects.create()
+        r = R.objects.create()
+        
+        def fill():            
+            for extra_kwargs in extra:
+                kwargs = {'m': m, 'r': r}
+                kwargs.update(extra_kwargs)
+                through.objects.create(**kwargs)
+
+        manager = getattr(m, name)
+        reverse_manager = getattr(r, "%s_m_set" % name)
+
+        fill()
+        manager.remove(r)
+        self.failIf(manager.exists())
+        self.failIf(reverse_manager.exists())
+        
+        fill()
+        reverse_manager.remove(m)
+        self.failIf(reverse_manager.exists())
+        self.failIf(manager.exists())
+        
+    def _test_managers(self, name, can_remove=False, can_add=False, extra=()):
+        if can_add:
+            self.assert_can_add(name)
+        else:
+            self.assert_cannot_add(name)
+        if can_remove:
+            self.assert_can_remove(name, extra)
+        else:
+            self.assert_cannot_remove(name)
+
+    def test_default(self):
+        self._test_managers('default', can_add=True, can_remove=True, extra=[{}])
+        
+    def test_default_cannot_remove(self):
+        self._test_managers('default_cannot_remove', can_add=True, can_remove=False)
+        
+    def test_default_cannot_add(self):
+        self._test_managers('default_cannot_add', can_add=False, can_remove=True, extra=[{}])
+
+    def test_through_default(self):
+        self._test_managers('through_default', can_add=False, can_remove=False)
+        
+    def test_through_auto(self):
+        self._test_managers('through_auto', can_add=True, can_remove=False)
+        
+    def test_through_ut(self):
+        self._test_managers('through_ut', can_add=False, can_remove=True, extra=[{'extra': 'foo'}])
+        
+    def test_through_can_add(self):
+        self._test_managers('through_can_add', can_add=True, can_remove=False)
+        
+    def test_through_can_remove(self):
+        self._test_managers('through_can_remove', can_add=False, can_remove=True, extra=[{'extra': 'foo'}, {'extra': 'bar'}])
+        
+ No newline at end of file

tests/regressiontests/m2m_through_regress/models.py

@@ -80 +80 @@
 >>> roll.members.all()
 [<Person: Bob>]
 
-# Error messages use the model name, not repr of the class name
->>> bob.group_set = []
-Traceback (most recent call last):
-...
-AttributeError: Cannot set values on a ManyToManyField which specifies an intermediary model.  Use m2m_through_regress.Membership's Manager instead.
-
->>> roll.members = []
-Traceback (most recent call last):
-...
-AttributeError: Cannot set values on a ManyToManyField which specifies an intermediary model.  Use m2m_through_regress.Membership's Manager instead.
-
->>> rock.members.create(name='Anne')
-Traceback (most recent call last):
-...
-AttributeError: Cannot use create() on a ManyToManyField which specifies an intermediary model.  Use m2m_through_regress.Membership's Manager instead.
-
->>> bob.group_set.create(name='Funk')
-Traceback (most recent call last):
-...
-AttributeError: Cannot use create() on a ManyToManyField which specifies an intermediary model.  Use m2m_through_regress.Membership's Manager instead.
-
 # Now test that the intermediate with a relationship outside
 # the current app (i.e., UserMembership) workds
 >>> UserMembership.objects.create(user=frank, group=rock)

django/db/models/fields/related.py

@@ -418 +418 @@
 
         return manager
 
-def create_many_related_manager(superclass, rel=False):
+def create_many_related_manager(superclass, field):
     """Creates a manager that subclasses 'superclass' (which is a Manager)
     and adds behavior for many-to-many related objects."""
-    through = rel.through
+    through = field.rel.through
+    can_add = field.can_add()
+    can_remove = field.can_remove()
+    
     class ManyRelatedManager(superclass):
         def __init__(self, model=None, core_filters=None, instance=None, symmetrical=None,
                 join_table=None, source_field_name=None, target_field_name=None):
@@ -440 +443 @@
         def get_query_set(self):
             return superclass.get_query_set(self).using(self.instance._state.db)._next_is_sticky().filter(**(self.core_filters))
 
-        # If the ManyToMany relation has an intermediary model,
-        # the add and remove methods do not exist.
-        if rel.through._meta.auto_created:
+        if can_add:
             def add(self, *objs):
                 self._add_items(self.source_field_name, self.target_field_name, *objs)
 
@@ -451 +452 @@
                     self._add_items(self.target_field_name, self.source_field_name, *objs)
             add.alters_data = True
 
+        if can_remove:
             def remove(self, *objs):
                 self._remove_items(self.source_field_name, self.target_field_name, *objs)
 
@@ -470 +472 @@
         def create(self, **kwargs):
             # This check needs to be done here, since we can't later remove this
             # from the method lookup table, as we do with add and remove.
-            if not rel.through._meta.auto_created:
+            if not can_add:
                 opts = through._meta
-                raise AttributeError, "Cannot use create() on a ManyToManyField which specifies an intermediary model. Use %s.%s's Manager instead." % (opts.app_label, opts.object_name)
+                raise AttributeError("Cannot use create() on this ManyToManyField. Use %s.%s's Manager instead." % (opts.app_label, opts.object_name))
             new_obj = super(ManyRelatedManager, self).using(self.instance._state.db).create(**kwargs)
             self.add(new_obj)
             return new_obj
         create.alters_data = True
 
         def get_or_create(self, **kwargs):
+            # This check needs to be done here, since we can't later remove this
+            # from the method lookup table, as we do with add and remove.
+            if not can_add:
+                opts = through._meta
+                raise AttributeError, "Cannot use get_or_create() on this ManyToManyField. Use %s.%s's Manager instead." % (opts.app_label, opts.object_name)
             obj, created = \
                     super(ManyRelatedManager, self).using(self.instance._state.db).get_or_create(**kwargs)
             # We only need to add() if created because if we got an object back
@@ -568 +575 @@
         # model's default manager.
         rel_model = self.related.model
         superclass = rel_model._default_manager.__class__
-        RelatedManager = create_many_related_manager(superclass, self.related.field.rel)
+        RelatedManager = create_many_related_manager(superclass, self.related.field)
 
         manager = RelatedManager(
             model=rel_model,
@@ -585 +592 @@
         if instance is None:
             raise AttributeError, "Manager must be accessed via instance"
 
-        if not self.related.field.rel.through._meta.auto_created:
+        if not self.related.field.can_add():
             opts = self.related.field.rel.through._meta
-            raise AttributeError, "Cannot set values on a ManyToManyField which specifies an intermediary model. Use %s.%s's Manager instead." % (opts.app_label, opts.object_name)
+            raise AttributeError, "Cannot set values on this ManyToManyField. Use %s.%s's Manager instead." % (opts.app_label, opts.object_name)
 
         manager = self.__get__(instance)
         manager.clear()
@@ -618 +625 @@
         # model's default manager.
         rel_model=self.field.rel.to
         superclass = rel_model._default_manager.__class__
-        RelatedManager = create_many_related_manager(superclass, self.field.rel)
+        RelatedManager = create_many_related_manager(superclass, self.field)
 
         manager = RelatedManager(
             model=rel_model,
@@ -635 +642 @@
         if instance is None:
             raise AttributeError, "Manager must be accessed via instance"
 
-        if not self.field.rel.through._meta.auto_created:
+        if not self.field.can_add():
             opts = self.field.rel.through._meta
-            raise AttributeError, "Cannot set values on a ManyToManyField which specifies an intermediary model.  Use %s.%s's Manager instead." % (opts.app_label, opts.object_name)
+            raise AttributeError, "Cannot set values a this ManyToManyField.  Use %s.%s's Manager instead." % (opts.app_label, opts.object_name)
 
         manager = self.__get__(instance)
         manager.clear()
@@ -883 +890 @@
         self.db_table = kwargs.pop('db_table', None)
         if kwargs['rel'].through is not None:
             assert self.db_table is None, "Cannot specify a db_table if an intermediary model is used."
+            
+        self._can_add = kwargs.pop('can_add', None)
+        self._can_remove = kwargs.pop('can_remove', None)
 
         Field.__init__(self, **kwargs)
 
@@ -901 +911 @@
         else:
             return util.truncate_name('%s_%s' % (opts.db_table, self.name),
                                       connection.ops.max_name_length())
-
-    def _get_m2m_attr(self, related, attr):
-        "Function that can be curried to provide the source column name for the m2m table"
-        cache_attr = '_m2m_%s_cache' % attr
+                                      
+    def _get_intermediary_fields(self, related):
+        cache_attr = '_m2m_intermediary_fields_cache'
         if hasattr(self, cache_attr):
             return getattr(self, cache_attr)
+            
+        candidates = []
+        related_candidates = []
+        auto_add = True
         for f in self.rel.through._meta.fields:
-            if hasattr(f,'rel') and f.rel and f.rel.to == related.model:
-                setattr(self, cache_attr, getattr(f, attr))
-                return getattr(self, cache_attr)
+            if hasattr(f,'rel') and f.rel:
+                if f.rel.to == related.model:
+                    candidates.append(f)
+                    continue
+                elif f.rel.to == related.parent_model:
+                    related_candidates.append(f)
+                    continue
+            if isinstance(f, AutoField) or f.null or f.has_default():
+                continue
+            elif getattr(f, 'auto_now_add', False) or getattr(f, 'auto_now', False):
+                continue
+            else:
+                auto_add = False
+        if related.model == related.parent_model:
+            # m2m to self
+            assert len(candidates) == 2, "There are too many ForeignKeys to %s" % related.model
+            field, related_field = candidates
+        else:
+            assert len(candidates) == 1, "There are no ForeignKeys to %s" % related.model
+            assert len(related_candidates) == 1, "There are no ForeignKeys to %s" % related.parent_model
+            # TODO: intelligently pick a candidate if there is more than one. For now, just use the first.
+            field, related_field = candidates[0], related_candidates[0]
 
+        if self._can_add is None:
+            self._can_add = auto_add
+
+        if self._can_remove is None:
+            self._can_remove = False
+            unique_together = [frozenset(ut) for ut in self.rel.through._meta.unique_together]                
+            names = frozenset([field.name, related_field.name])
+            for ut in unique_together:
+                if names <= ut:
+                    self._can_remove = True
+                    break
+        
+        setattr(self, cache_attr, (field, related_field))
+        return (field, related_field)
+        
+    def _get_can_add(self, related):
+        if self._can_add is None:
+            self._get_intermediary_fields(related)
+        return self._can_add
+        
+    def _get_can_remove(self, related):
+        if self._can_remove is None:
+            self._get_intermediary_fields(related)
+        return self._can_remove
+
+    def _get_m2m_attr(self, related, attr):
+        "Function that can be curried to provide a source field attribute"
+        cache_attr = '_m2m_%s_cache' % attr
+        if not hasattr(self, cache_attr):
+            field, _ = self._get_intermediary_fields(related)
+            setattr(self, cache_attr, getattr(field, attr))
+        return getattr(self, cache_attr)
+
     def _get_m2m_reverse_attr(self, related, attr):
-        "Function that can be curried to provide the related column name for the m2m table"
+        "Function that can be curried to provide a related field attribute"
         cache_attr = '_m2m_reverse_%s_cache' % attr
-        if hasattr(self, cache_attr):
-            return getattr(self, cache_attr)
-        found = False
-        for f in self.rel.through._meta.fields:
-            if hasattr(f,'rel') and f.rel and f.rel.to == related.parent_model:
-                if related.model == related.parent_model:
-                    # If this is an m2m-intermediate to self,
-                    # the first foreign key you find will be
-                    # the source column. Keep searching for
-                    # the second foreign key.
-                    if found:
-                        setattr(self, cache_attr, getattr(f, attr))
-                        break
-                    else:
-                        found = True
-                else:
-                    setattr(self, cache_attr, getattr(f, attr))
-                    break
+        if not hasattr(self, cache_attr):
+            _, related_field = self._get_intermediary_fields(related)
+            setattr(self, cache_attr, getattr(related_field, attr))            
         return getattr(self, cache_attr)
 
     def isValidIDList(self, field_data, all_data):
@@ -1017 +1067 @@
 
         self.m2m_field_name = curry(self._get_m2m_attr, related, 'name')
         self.m2m_reverse_field_name = curry(self._get_m2m_reverse_attr, related, 'name')
+        
+        self.can_add = curry(self._get_can_add, related)
+        self.can_remove = curry(self._get_can_remove, related)
 
     def set_attributes_from_rel(self):
         pass