fix and test impl

Author: bckohan

src/polymorphic/managers.py

@@ -2,8 +2,6 @@
 The manager class for use in the models.
 """
 
-import inspect
-
 from django.contrib.contenttypes.models import ContentType
 from django.db import DEFAULT_DB_ALIAS, models
 
@@ -63,33 +61,33 @@ def create_from_super(self, obj, **kwargs):
 
         returns obj as an instance of cls.
         """
-        cls = self.model
+        from .models import PolymorphicModel
+
+        # ensure we have the most derived real instance
+        if isinstance(obj, PolymorphicModel):
+            obj = obj.get_real_instance()
+
+        parent_ptr = self.model._meta.parents.get(type(obj), None)
 
-        scls = inspect.getmro(cls)[1]
-        if scls is not type(obj):
+        if not parent_ptr:
             raise TypeError(
                 "create_from_super can only be used if obj is one level of inheritance up from cls"
             )
-
-        parent_link_field = None
-        for parent, field in cls._meta.parents.items():
-            if parent is scls:
-                parent_link_field = field
-                break
-        if parent_link_field is None:
-            raise TypeError(f"Could not find parent link field for {scls.__name__}")
-        kwargs[parent_link_field.get_attname()] = obj.id
+        kwargs[parent_ptr.get_attname()] = obj.pk
 
         # create the new base class with only fields that apply to  it.
-        nobj = cls(**kwargs)
-        nobj.save_base(raw=True)
+        ctype = ContentType.objects.db_manager(
+            using=(obj._state.db or DEFAULT_DB_ALIAS)
+        ).get_for_model(self.model)
+        nobj = self.model(**kwargs, polymorphic_ctype=ctype)
+        nobj.save_base(raw=True, using=obj._state.db or DEFAULT_DB_ALIAS, force_insert=True)
         # force update the content type, but first we need to
         # retrieve a clean copy from the db to fill in the null
         # fields otherwise they would be overwritten.
-        nobj = obj.__class__.objects.using(obj._state.db or DEFAULT_DB_ALIAS).get(pk=obj.pk)
-        nobj.polymorphic_ctype = ContentType.objects.db_manager(
-            using=(obj._state.db or DEFAULT_DB_ALIAS)
-        ).get_for_model(cls)
-        nobj.save()
+        if isinstance(obj, PolymorphicModel):
+            parent = obj.__class__.objects.using(obj._state.db or DEFAULT_DB_ALIAS).get(pk=obj.pk)
+            parent.polymorphic_ctype = ctype
+            parent.save()
 
-        return nobj.get_real_instance()  # cast to cls
+        nobj.refresh_from_db()  # cast to cls
+        return nobj

src/polymorphic/tests/migrations/0001_initial.py

@@ -1,4 +1,4 @@
-# Generated by Django 4.2 on 2025-12-12 18:24
+# Generated by Django 4.2 on 2025-12-13 21:30
 
 from django.conf import settings
 from django.db import migrations, models

src/polymorphic/tests/test_multidb.py

@@ -137,3 +137,142 @@ def run():
 
         # Ensure no queries are made using the default database.
         self.assertNumQueries(0, run)
+
+    def test_create_from_super(self):
+        # run create test 3 times because initial implementation
+        # would fail after first success.
+        from polymorphic.tests.models import (
+            NormalBase,
+            NormalExtension,
+            PolyExtension,
+            PolyExtChild,
+        )
+
+        nb = NormalBase.objects.db_manager("secondary").create(nb_field=1)
+        ne = NormalExtension.objects.db_manager("secondary").create(nb_field=2, ne_field="ne2")
+
+        with self.assertRaises(TypeError):
+            PolyExtension.objects.db_manager("secondary").create_from_super(nb, poly_ext_field=3)
+
+        pe = PolyExtension.objects.db_manager("secondary").create_from_super(ne, poly_ext_field=3)
+
+        ne.refresh_from_db()
+        self.assertEqual(type(ne), NormalExtension)
+        self.assertEqual(type(pe), PolyExtension)
+        self.assertEqual(pe.pk, ne.pk)
+
+        self.assertEqual(pe.nb_field, 2)
+        self.assertEqual(pe.ne_field, "ne2")
+        self.assertEqual(pe.poly_ext_field, 3)
+        pe.refresh_from_db()
+        self.assertEqual(pe.nb_field, 2)
+        self.assertEqual(pe.ne_field, "ne2")
+        self.assertEqual(pe.poly_ext_field, 3)
+
+        pc = PolyExtChild.objects.db_manager("secondary").create_from_super(
+            pe, poly_child_field="pcf6"
+        )
+
+        pe.refresh_from_db()
+        ne.refresh_from_db()
+        self.assertEqual(type(ne), NormalExtension)
+        self.assertEqual(type(pe), PolyExtension)
+        self.assertEqual(pe.pk, ne.pk)
+        self.assertEqual(pe.pk, pc.pk)
+
+        self.assertEqual(pc.nb_field, 2)
+        self.assertEqual(pc.ne_field, "ne2")
+        self.assertEqual(pc.poly_ext_field, 3)
+        pc.refresh_from_db()
+        self.assertEqual(pc.nb_field, 2)
+        self.assertEqual(pc.ne_field, "ne2")
+        self.assertEqual(pc.poly_ext_field, 3)
+        self.assertEqual(pc.poly_child_field, "pcf6")
+
+        self.assertEqual(
+            pe.polymorphic_ctype,
+            ContentType.objects.db_manager("secondary").get_for_model(PolyExtChild),
+        )
+        self.assertEqual(
+            pc.polymorphic_ctype,
+            ContentType.objects.db_manager("secondary").get_for_model(PolyExtChild),
+        )
+
+        self.assertEqual(set(PolyExtension.objects.db_manager("secondary").all()), {pc})
+
+        a1 = Model2A.objects.db_manager("secondary").create(field1="A1a")
+        a2 = Model2A.objects.db_manager("secondary").create(field1="A1b")
+
+        b1 = Model2B.objects.db_manager("secondary").create(field1="B1a", field2="B2a")
+        b2 = Model2B.objects.db_manager("secondary").create(field1="B1b", field2="B2b")
+
+        c1 = Model2C.objects.db_manager("secondary").create(
+            field1="C1a", field2="C2a", field3="C3a"
+        )
+        c2 = Model2C.objects.db_manager("secondary").create(
+            field1="C1b", field2="C2b", field3="C3b"
+        )
+
+        d1 = Model2D.objects.db_manager("secondary").create(
+            field1="D1a", field2="D2a", field3="D3a", field4="D4a"
+        )
+        d2 = Model2D.objects.db_manager("secondary").create(
+            field1="D1b", field2="D2b", field3="D3b", field4="D4b"
+        )
+
+        with self.assertRaises(TypeError):
+            Model2D.objects.db_manager("secondary").create_from_super(
+                b1, field3="D3x", field4="D4x"
+            )
+
+        b1_of_c = Model2B.objects.db_manager("secondary").non_polymorphic().get(pk=c1.pk)
+        with self.assertRaises(TypeError):
+            Model2C.objects.db_manager("secondary").create_from_super(b1_of_c, field3="C3x")
+
+        self.assertEqual(
+            c1.polymorphic_ctype,
+            ContentType.objects.db_manager("secondary").get_for_model(Model2C),
+        )
+        dfs1 = Model2D.objects.db_manager("secondary").create_from_super(b1_of_c, field4="D4x")
+        self.assertEqual(type(dfs1), Model2D)
+        self.assertEqual(dfs1.pk, c1.pk)
+        self.assertEqual(dfs1.field1, "C1a")
+        self.assertEqual(dfs1.field2, "C2a")
+        self.assertEqual(dfs1.field3, "C3a")
+        self.assertEqual(dfs1.field4, "D4x")
+        self.assertEqual(
+            dfs1.polymorphic_ctype,
+            ContentType.objects.db_manager("secondary").get_for_model(Model2D),
+        )
+        c1.refresh_from_db()
+        self.assertEqual(
+            c1.polymorphic_ctype,
+            ContentType.objects.db_manager("secondary").get_for_model(Model2D),
+        )
+
+        self.assertEqual(
+            b2.polymorphic_ctype,
+            ContentType.objects.db_manager("secondary").get_for_model(Model2B),
+        )
+        cfs1 = Model2C.objects.db_manager("secondary").create_from_super(b2, field3="C3y")
+        self.assertEqual(type(cfs1), Model2C)
+        self.assertEqual(cfs1.pk, b2.pk)
+        self.assertEqual(cfs1.field1, "B1b")
+        self.assertEqual(cfs1.field2, "B2b")
+        self.assertEqual(cfs1.field3, "C3y")
+        b2.refresh_from_db()
+        self.assertEqual(
+            b2.polymorphic_ctype,
+            ContentType.objects.db_manager("secondary").get_for_model(Model2C),
+        )
+        self.assertEqual(
+            cfs1.polymorphic_ctype,
+            ContentType.objects.db_manager("secondary").get_for_model(Model2C),
+        )
+
+        self.assertEqual(
+            set(Model2A.objects.db_manager("secondary").all()),
+            {a1, a2, b1, dfs1, cfs1, c2, d1, d2},
+        )
+
+        self.assertEqual(Model2A.objects.count(), 0)

src/polymorphic/tests/test_orm.py

@@ -11,7 +11,7 @@
 
 from polymorphic import query_translate
 from polymorphic.managers import PolymorphicManager
-from polymorphic.models import PolymorphicTypeInvalid, PolymorphicTypeUndefined
+from polymorphic.models import PolymorphicModel, PolymorphicTypeInvalid, PolymorphicTypeUndefined
 from polymorphic.tests.models import (
     ArtProject,
     Base,
@@ -24,6 +24,7 @@
     CustomPkBase,
     CustomPkInherit,
     Enhance_Base,
+    Enhance_Plain,
     Enhance_Inherit,
     InlineParent,
     InlineModelA,
@@ -1745,3 +1746,101 @@ def test_polymorphic_extension(self):
         }
         assert set(PolyExtension.objects.all()) == {poly_ext, child_ext}
         assert set(PolyExtChild.objects.all()) == {child_ext}
+
+    def test_create_from_super(self):
+        # run create test 3 times because initial implementation
+        # would fail after first success.
+        from polymorphic.tests.models import (
+            NormalBase,
+            NormalExtension,
+            PolyExtension,
+            PolyExtChild,
+        )
+
+        nb = NormalBase.objects.create(nb_field=1)
+        ne = NormalExtension.objects.create(nb_field=2, ne_field="ne2")
+
+        with self.assertRaises(TypeError):
+            PolyExtension.objects.create_from_super(nb, poly_ext_field=3)
+
+        pe = PolyExtension.objects.create_from_super(ne, poly_ext_field=3)
+
+        ne.refresh_from_db()
+        self.assertEqual(type(ne), NormalExtension)
+        self.assertEqual(type(pe), PolyExtension)
+        self.assertEqual(pe.pk, ne.pk)
+
+        self.assertEqual(pe.nb_field, 2)
+        self.assertEqual(pe.ne_field, "ne2")
+        self.assertEqual(pe.poly_ext_field, 3)
+        pe.refresh_from_db()
+        self.assertEqual(pe.nb_field, 2)
+        self.assertEqual(pe.ne_field, "ne2")
+        self.assertEqual(pe.poly_ext_field, 3)
+
+        pc = PolyExtChild.objects.create_from_super(pe, poly_child_field="pcf6")
+
+        pe.refresh_from_db()
+        ne.refresh_from_db()
+        self.assertEqual(type(ne), NormalExtension)
+        self.assertEqual(type(pe), PolyExtension)
+        self.assertEqual(pe.pk, ne.pk)
+        self.assertEqual(pe.pk, pc.pk)
+
+        self.assertEqual(pc.nb_field, 2)
+        self.assertEqual(pc.ne_field, "ne2")
+        self.assertEqual(pc.poly_ext_field, 3)
+        pc.refresh_from_db()
+        self.assertEqual(pc.nb_field, 2)
+        self.assertEqual(pc.ne_field, "ne2")
+        self.assertEqual(pc.poly_ext_field, 3)
+        self.assertEqual(pc.poly_child_field, "pcf6")
+
+        self.assertEqual(pe.polymorphic_ctype, ContentType.objects.get_for_model(PolyExtChild))
+        self.assertEqual(pc.polymorphic_ctype, ContentType.objects.get_for_model(PolyExtChild))
+
+        self.assertEqual(set(PolyExtension.objects.all()), {pc})
+
+        a1 = Model2A.objects.create(field1="A1a")
+        a2 = Model2A.objects.create(field1="A1b")
+
+        b1 = Model2B.objects.create(field1="B1a", field2="B2a")
+        b2 = Model2B.objects.create(field1="B1b", field2="B2b")
+
+        c1 = Model2C.objects.create(field1="C1a", field2="C2a", field3="C3a")
+        c2 = Model2C.objects.create(field1="C1b", field2="C2b", field3="C3b")
+
+        d1 = Model2D.objects.create(field1="D1a", field2="D2a", field3="D3a", field4="D4a")
+        d2 = Model2D.objects.create(field1="D1b", field2="D2b", field3="D3b", field4="D4b")
+
+        with self.assertRaises(TypeError):
+            Model2D.objects.create_from_super(b1, field3="D3x", field4="D4x")
+
+        b1_of_c = Model2B.objects.non_polymorphic().get(pk=c1.pk)
+        with self.assertRaises(TypeError):
+            Model2C.objects.create_from_super(b1_of_c, field3="C3x")
+
+        self.assertEqual(c1.polymorphic_ctype, ContentType.objects.get_for_model(Model2C))
+        dfs1 = Model2D.objects.create_from_super(b1_of_c, field4="D4x")
+        self.assertEqual(type(dfs1), Model2D)
+        self.assertEqual(dfs1.pk, c1.pk)
+        self.assertEqual(dfs1.field1, "C1a")
+        self.assertEqual(dfs1.field2, "C2a")
+        self.assertEqual(dfs1.field3, "C3a")
+        self.assertEqual(dfs1.field4, "D4x")
+        self.assertEqual(dfs1.polymorphic_ctype, ContentType.objects.get_for_model(Model2D))
+        c1.refresh_from_db()
+        self.assertEqual(c1.polymorphic_ctype, ContentType.objects.get_for_model(Model2D))
+
+        self.assertEqual(b2.polymorphic_ctype, ContentType.objects.get_for_model(Model2B))
+        cfs1 = Model2C.objects.create_from_super(b2, field3="C3y")
+        self.assertEqual(type(cfs1), Model2C)
+        self.assertEqual(cfs1.pk, b2.pk)
+        self.assertEqual(cfs1.field1, "B1b")
+        self.assertEqual(cfs1.field2, "B2b")
+        self.assertEqual(cfs1.field3, "C3y")
+        b2.refresh_from_db()
+        self.assertEqual(b2.polymorphic_ctype, ContentType.objects.get_for_model(Model2C))
+        self.assertEqual(cfs1.polymorphic_ctype, ContentType.objects.get_for_model(Model2C))
+
+        self.assertEqual(set(Model2A.objects.all()), {a1, a2, b1, dfs1, cfs1, c2, d1, d2})