Allow ManyToManyFields that target proxy models with a through table.

[Collin Anderson]

I'm running into an issue with using ManyToManyFields and Proxy models when the through models reference Proxy classes of the original model.

​https://github.com/django/django/pull/10591

[Tim Graham]

My comment requesting more carefully targeted tests isn't addressed.

[Collin Anderson]

Ok, I just added some tests.

[Collin Anderson]

Clarify title/description a little.

[jaehyuck.sa.dev]

My GitHub account is currently suspended, so PR ​​https://github.com/django/django/pull/20554 is gone. I also used GitHub to log in to Trac, so I can't access that account either.

I'm posting this from a new Trac account, and I can't attach files here, so I'm just pasting the diff below. This is the patch from ​https://github.com/django/django/pull/20554 rebased onto the current main (fb1137658ff).

There are still two review comments from the original PR that I haven't addressed in this diff:

• ontowhee asked whether _get_m2m_attr() should use _get_concrete_model() on both sides for consistency, and the same for _get_m2m_reverse_attr().
• raffaellasuardini pointed out that the tests cover remove(), but not clear().

David kindly offered to open a PR with this on GitHub, so feel free to take it from here and address the two review comments above as well. Thanks!

Here's the diff:

django/db/models/fields/related.py

@@ -1645 +1645 @@
                     )
             relationship_model_name = self.remote_field.through._meta.object_name
             self_referential = from_model == to_model
+
+            def _get_concrete_model(model):
+                if model is None or isinstance(model, str):
+                    return model
+                return model._meta.concrete_model
+
             # Count foreign keys in intermediate model
             if self_referential:
                 seen_self = sum(
-                    from_model == getattr(field.remote_field, "model", None)
+                    _get_concrete_model(from_model)
+                    == _get_concrete_model(getattr(field.remote_field, "model", None))
                     for field in self.remote_field.through._meta.fields
                 )
 
@@ -1671 +1678 @@
                     )
 
             else:
-                # Count foreign keys in relationship model
                 seen_from = sum(
-                    from_model == getattr(field.remote_field, "model", None)
+                    _get_concrete_model(from_model)
+                    == _get_concrete_model(getattr(field.remote_field, "model", None))
                     for field in self.remote_field.through._meta.fields
                 )
                 seen_to = sum(
-                    to_model == getattr(field.remote_field, "model", None)
+                    _get_concrete_model(to_model)
+                    == _get_concrete_model(getattr(field.remote_field, "model", None))
                     for field in self.remote_field.through._meta.fields
                 )
 
@@ -1785 +1793 @@
                 ):
                     possible_field_names = []
                     for f in through._meta.fields:
-                        if (
-                            hasattr(f, "remote_field")
-                            and getattr(f.remote_field, "model", None) == related_model
-                        ):
+                        if hasattr(f, "remote_field") and _get_concrete_model(
+                            getattr(f.remote_field, "model", None)
+                        ) == _get_concrete_model(related_model):
                             possible_field_names.append(f.name)
+                    related_model_name = (
+                        related_model
+                        if isinstance(related_model, str)
+                        else related_model._meta.object_name
+                    )
                     if possible_field_names:
                         hint = (
                             "Did you mean one of the following foreign keys to '%s': "
                             "%s?"
-                            % (
-                                related_model._meta.object_name,
-                                ", ".join(possible_field_names),
-                            )
+                            % (related_model_name, ", ".join(possible_field_names))
                         )
                     else:
                         hint = None
@@ -1817 +1826 @@
                     else:
                         if not (
                             hasattr(field, "remote_field")
-                            and getattr(field.remote_field, "model", None)
-                            == related_model
-                        ):
-                            related_object_name = (
-                                related_model
-                                if isinstance(related_model, str)
-                                else related_model._meta.object_name
+                            and _get_concrete_model(
+                                getattr(field.remote_field, "model", None)
                             )
+                            == _get_concrete_model(related_model)
+                        ):
                             errors.append(
                                 checks.Error(
                                     "'%s.%s' is not a foreign key to '%s'."
                                     % (
                                         through._meta.object_name,
                                         field_name,
-                                        related_object_name,
+                                        related_model_name,
                                     ),
                                     hint=hint,
                                     obj=self,
@@ -2012 +2018 @@
         for f in self.remote_field.through._meta.fields:
             if (
                 f.is_relation
-                and f.remote_field.model == related.related_model
+                and f.remote_field.model._meta.concrete_model
+                == related.related_model._meta.concrete_model
                 and (link_field_name is None or link_field_name == f.name)
             ):
                 setattr(self, cache_attr, getattr(f, attr))
@@ -2032 +2039 @@
         else:
             link_field_name = None
         for f in self.remote_field.through._meta.fields:
-            if f.is_relation and f.remote_field.model == related.model:
+            if (
+                f.is_relation
+                and f.remote_field.model._meta.concrete_model
+                == related.model._meta.concrete_model
+            ):
                 if link_field_name is None and related.related_model == related.model:
                     # If this is an m2m-intermediate to self,
                     # the first foreign key you find will be

tests/m2m_through/models.py

@@ -26 +26 @@
         through="TestNoDefaultsOrNulls",
         related_name="testnodefaultsnonulls",
     )
+    proxy_members = models.ManyToManyField(
+        Person,
+        through="ProxyMembership",
+        through_fields=("group", "proxy_person"),
+        related_name="proxy_groups",
+    )
+    double_proxy_members = models.ManyToManyField(
+        "ProxyPerson",
+        through="DoubleProxyMembership",
+        through_fields=("proxy_group", "proxy_person"),
+        related_name="double_proxy_groups",
+    )
 
     class Meta:
         ordering = ("name",)
@@ -155 +167 @@
 class RecipeIngredient(models.Model):
     ingredient = models.ForeignKey(Ingredient, models.CASCADE, to_field="iname")
     recipe = models.ForeignKey(Recipe, models.CASCADE, to_field="rname")
+
+
+class ProxyPerson(Person):
+    class Meta:
+        proxy = True
+
+
+class ProxyGroup(Group):
+    class Meta:
+        proxy = True
+
+
+class ProxyMembership(models.Model):
+    proxy_person = models.ForeignKey(ProxyPerson, models.CASCADE)
+    group = models.ForeignKey(Group, models.CASCADE)
+    joined = models.DateTimeField(default=datetime.now)
+
+
+class DoubleProxyMembership(models.Model):
+    proxy_person = models.ForeignKey(ProxyPerson, models.CASCADE)
+    proxy_group = models.ForeignKey(ProxyGroup, models.CASCADE)

tests/m2m_through/tests.py

@@ -2 +2 @@
 from operator import attrgetter
 
 from django.db import IntegrityError
+from django.forms.models import inlineformset_factory
 from django.test import TestCase
 
 from .models import (
     CustomMembership,
+    DoubleProxyMembership,
     Employee,
     Event,
     Friendship,
@@ -16 +18 @@
     Person,
     PersonChild,
     PersonSelfRefM2M,
+    ProxyGroup,
+    ProxyMembership,
+    ProxyPerson,
     Recipe,
     RecipeIngredient,
     Relationship,
@@ -542 +547 @@
     def test_exists(self):
         self.assertTrue(self.curry.ingredients.exists())
         self.assertTrue(self.tomato.recipes.exists())
+
+
+class M2mThroughProxyTests(TestCase):
+    @classmethod
+    def setUpTestData(cls):
+        cls.person = Person.objects.create(name="Alice")
+        cls.proxy_person = ProxyPerson.objects.create(name="Bob")
+        cls.group = Group.objects.create(name="Team")
+        cls.proxy_group = ProxyGroup.objects.create(name="Proxy Team")
+
+    def test_forward(self):
+        test_cases = [
+            (
+                ProxyMembership,
+                {"proxy_person": self.person, "group": self.group},
+                self.group,
+                "proxy_members",
+                [self.person],
+            ),
+            (
+                DoubleProxyMembership,
+                {"proxy_person": self.proxy_person, "proxy_group": self.proxy_group},
+                self.proxy_group,
+                "double_proxy_members",
+                [self.proxy_person],
+            ),
+        ]
+        for model, create_kwargs, parent, attr, expected in test_cases:
+            with self.subTest(model=model.__name__):
+                model.objects.create(**create_kwargs)
+                self.assertSequenceEqual(list(getattr(parent, attr).all()), expected)
+
+    def test_reverse(self):
+        test_cases = [
+            (
+                ProxyMembership,
+                {"proxy_person": self.person, "group": self.group},
+                self.person,
+                "proxy_groups",
+                [self.group],
+            ),
+            (
+                DoubleProxyMembership,
+                {"proxy_person": self.proxy_person, "proxy_group": self.proxy_group},
+                self.proxy_person,
+                "double_proxy_groups",
+                [self.proxy_group],
+            ),
+        ]
+        for model, create_kwargs, related, attr, expected in test_cases:
+            with self.subTest(model=model.__name__):
+                model.objects.create(**create_kwargs)
+                self.assertSequenceEqual(list(getattr(related, attr).all()), expected)
+
+    def test_add(self):
+        test_cases = [
+            (
+                self.group,
+                "proxy_members",
+                self.person,
+                ProxyMembership,
+                {"proxy_person": self.person, "group": self.group},
+            ),
+            (
+                self.proxy_group,
+                "double_proxy_members",
+                self.proxy_person,
+                DoubleProxyMembership,
+                {"proxy_person": self.proxy_person, "proxy_group": self.proxy_group},
+            ),
+        ]
+        for parent, attr, member, model, filter_kwargs in test_cases:
+            with self.subTest(model=model.__name__):
+                getattr(parent, attr).add(member)
+                self.assertTrue(model.objects.filter(**filter_kwargs).exists())
+
+    def test_remove(self):
+        test_cases = [
+            (
+                ProxyMembership,
+                {"proxy_person": self.person, "group": self.group},
+                self.group,
+                "proxy_members",
+                self.person,
+            ),
+            (
+                DoubleProxyMembership,
+                {"proxy_person": self.proxy_person, "proxy_group": self.proxy_group},
+                self.proxy_group,
+                "double_proxy_members",
+                self.proxy_person,
+            ),
+        ]
+        for model, create_kwargs, parent, attr, member in test_cases:
+            with self.subTest(model=model.__name__):
+                model.objects.create(**create_kwargs)
+                getattr(parent, attr).remove(member)
+                self.assertFalse(model.objects.filter(**create_kwargs).exists())
+
+    def test_inlineformset(self):
+        test_cases = [
+            (Group, ProxyMembership, self.group),
+            (ProxyGroup, DoubleProxyMembership, self.proxy_group),
+        ]
+        for parent_model, through_model, instance in test_cases:
+            with self.subTest(through_model=through_model.__name__):
+                FormSet = inlineformset_factory(
+                    parent_model, through_model, fields=["proxy_person"]
+                )
+                formset = FormSet(instance=instance)
+                self.assertEqual(formset.model, through_model)