Method bulk_create

ormar/queryset/queryset.py:1365–1395

Performs a bulk create in one database session to speed up the process. Allows you to create multiple objects at once. A valid list of `Model` objects needs to be passed. Bulk operations do not send signals. :param objects: list of ormar models already in

(self, objects: list["T"])

        return instance

    async def bulk_create(self, objects: list["T"]) -> None:
        """
        Performs a bulk create in one database session to speed up the process.

        Allows you to create multiple objects at once.

        A valid list of `Model` objects needs to be passed.

        Bulk operations do not send signals.

        :param objects: list of ormar models already initialized and ready to save.
        :type objects: list[Model]
        """

        if not objects:
            raise ModelListEmptyError("Bulk create objects are empty!")

        ready_objects = []
        for i, obj in enumerate(objects):
            ready_objects.append(obj.prepare_model_to_save(obj.model_dump()))
            if i % 100 == 99:  # pragma: no cover
                await asyncio.sleep(0)

        # don't use execute_many, as in databases it's executed in a loop
        # instead of using execute_many from drivers
        expr = self.table.insert().values(ready_objects)
        async with self.model_config.database.get_query_executor() as executor:
            await executor.execute(expr)

        for obj in objects:
            obj.set_save_status(True)

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