manager.bulk_create(...) with heterogeneous list of models using InheritanceManager fails

[Iapetus-11]

Problem

Doing MyBaseModel.objects.bulk_create([MyModelA(), MyModelB()]) results in an AssertionError.

Traceback

/Users/miloi/.local/share/virtualenvs/my-api-vsQQamSD/lib/python3.10/site-packages/rest_framework/mixins.py:19: in create
    self.perform_create(serializer)
../../../myapi/particle/views/at_event_stream_views.py:28: in perform_create
    MyBaseModel.objects.bulk_create(new_events)
/Users/miloi/.local/share/virtualenvs/my-api-vsQQamSD/lib/python3.10/site-packages/django/db/models/manager.py:87: in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <InheritanceQuerySet []>
objs = [<MyModelA: MyModelA object (None)>, <MyModelB: MyModelB object (None)>, <MyModelA: MyModelA object (None)...odelA: MyModelA object (None)>, <MyModelB: MyModelB object (None)>, <MyModelA: MyModelA object (None)>, ...]
batch_size = None, ignore_conflicts = False, update_conflicts = False
update_fields = None, unique_fields = None

    def bulk_create(
        self,
        objs,
        batch_size=None,
        ignore_conflicts=False,
        update_conflicts=False,
        update_fields=None,
        unique_fields=None,
    ):
        """
        Insert each of the instances into the database. Do *not* call
        save() on each of the instances, do not send any pre/post_save
        signals, and do not set the primary key attribute if it is an
        autoincrement field (except if features.can_return_rows_from_bulk_insert=True).
        Multi-table models are not supported.
        """
        # When you bulk insert you don't get the primary keys back (if it's an
        # autoincrement, except if can_return_rows_from_bulk_insert=True), so
        # you can't insert into the child tables which references this. There
        # are two workarounds:
        # 1) This could be implemented if you didn't have an autoincrement pk
        # 2) You could do it by doing O(n) normal inserts into the parent
        #    tables to get the primary keys back and then doing a single bulk
        #    insert into the childmost table.
        # We currently set the primary keys on the objects when using
        # PostgreSQL via the RETURNING ID clause. It should be possible for
        # Oracle as well, but the semantics for extracting the primary keys is
        # trickier so it's not done yet.
        if batch_size is not None and batch_size <= 0:
            raise ValueError("Batch size must be a positive integer.")
        # Check that the parents share the same concrete model with the our
        # model to detect the inheritance pattern ConcreteGrandParent ->
        # MultiTableParent -> ProxyChild. Simply checking self.model._meta.proxy
        # would not identify that case as involving multiple tables.
        for parent in self.model._meta.get_parent_list():
            if parent._meta.concrete_model is not self.model._meta.concrete_model:
                raise ValueError("Can't bulk create a multi-table inherited model")
        if not objs:
            return objs
        opts = self.model._meta
        if unique_fields:
            # Primary key is allowed in unique_fields.
            unique_fields = [
                self.model._meta.get_field(opts.pk.name if name == "pk" else name)
                for name in unique_fields
            ]
        if update_fields:
            update_fields = [self.model._meta.get_field(name) for name in update_fields]
        on_conflict = self._check_bulk_create_options(
            ignore_conflicts,
            update_conflicts,
            update_fields,
            unique_fields,
        )
        self._for_write = True
        fields = opts.concrete_fields
        objs = list(objs)
        self._prepare_for_bulk_create(objs)
        with transaction.atomic(using=self.db, savepoint=False):
            objs_with_pk, objs_without_pk = partition(lambda o: o.pk is None, objs)
            if objs_with_pk:
                returned_columns = self._batched_insert(
                    objs_with_pk,
                    fields,
                    batch_size,
                    on_conflict=on_conflict,
                    update_fields=update_fields,
                    unique_fields=unique_fields,
                )
                for obj_with_pk, results in zip(objs_with_pk, returned_columns):
                    for result, field in zip(results, opts.db_returning_fields):
                        if field != opts.pk:
                            setattr(obj_with_pk, field.attname, result)
                for obj_with_pk in objs_with_pk:
                    obj_with_pk._state.adding = False
                    obj_with_pk._state.db = self.db
            if objs_without_pk:
                fields = [f for f in fields if not isinstance(f, AutoField)]
                returned_columns = self._batched_insert(
                    objs_without_pk,
                    fields,
                    batch_size,
                    on_conflict=on_conflict,
                    update_fields=update_fields,
                    unique_fields=unique_fields,
                )
                connection = connections[self.db]
                if (
                    connection.features.can_return_rows_from_bulk_insert
                    and on_conflict is None
                ):
>                   assert len(returned_columns) == len(objs_without_pk)
E                   AssertionError

/Users/miloi/.local/share/virtualenvs/my-api-vsQQamSD/lib/python3.10/site-packages/django/db/models/query.py:816: AssertionError
Destroying test database for alias 'default' ('test_my-api-test')...

Environment

• Django Model Utils version: 4.3.1
• Django version: 4.2.7
• Python version: 3.10.7
• Other libraries used, if any: djangorestframework, psycopg2, dj-database-url

Code examples

class MyBaseModel(Model):
    objects = InheritanceManager()


class MyModelA(MyBaseModel):
    ...


class MyModelB(MyBaseModel):
    ...


MyBaseModel.objects.bulk_create([MyModelA(), MyModelB()])