Method bulk_update

ormar/queryset/queryset.py:1397–1497

Performs bulk update in one database session to speed up the process. Allows you to update multiple instance at once. All `Models` passed need to have primary key column populated. You can also select which fields to update by passing `columns` list as a l

(  # noqa:  CCR001
        self, objects: list["T"], columns: Optional[list[str]] = None
    )

            obj.set_save_status(True)

    async def bulk_update(  # noqa:  CCR001
        self, objects: list["T"], columns: Optional[list[str]] = None
    ) -> None:
        """
        Performs bulk update in one database session to speed up the process.

        Allows you to update multiple instance at once.

        All `Models` passed need to have primary key column populated.

        You can also select which fields to update by passing `columns` list
        as a list of string names.

        Bulk operations do not send signals.

        :param objects: list of ormar models
        :type objects: list[Model]
        :param columns: list of columns to update
        :type columns: list[str]
        """
        if not objects:
            raise ModelListEmptyError("Bulk update objects are empty!")

        ready_objects = []
        pk_name = self.model_config.pkname
        if not columns:
            columns = list(
                self.model.extract_db_own_fields().union(
                    self.model.extract_related_names()
                )
            )

        if pk_name not in columns:
            columns.append(pk_name)

        onupdate_fields = self.model._onupdate_fields
        for field_name in onupdate_fields:
            if field_name not in columns:
                columns.append(field_name)

        columns = [self.model.get_column_alias(k) for k in columns]

        for i, obj in enumerate(objects):
            explicit_fields = obj.__setattr_fields__
            new_kwargs = obj.model_dump()
            if new_kwargs.get(pk_name) is None:
                raise ModelPersistenceError(
                    "You cannot update unsaved objects. "
                    f"{self.model.__name__} has to have {pk_name} filled."
                )
            new_kwargs = obj.populate_onupdate_value(
                new_kwargs, explicit_fields=explicit_fields
            )
            obj.update_from_dict(
                {
                    field_name: new_kwargs[field_name]
                    for field_name in onupdate_fields - explicit_fields
                }