mirrors/django
1
0
Fork 0
mirror of https://github.com/django/django.git synced 2025-10-31 09:41:08 +00:00
Code Issues 32 Releases Wiki Activity

Refs #36143 -- Tested bulk_batch_size limit for bulk_update and bulk_create.

Browse Source
This commit is contained in:
Sage Abdullah
2025-05-08 09:10:00 +02:00
committed by Sarah Boyce
parent 37f2a77c72
commit 38660a612c
2 changed files with 19 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
tests/bulk_create/tests.py
View File

@@ -293,6 +293,14 @@ class BulkCreateTests(TestCase):
with self.assertNumQueries(ceil(len(objs) / max_batch_size)): with self.assertNumQueries(ceil(len(objs) / max_batch_size)):
Country.objects.bulk_create(objs, batch_size=max_batch_size + 1) Country.objects.bulk_create(objs, batch_size=max_batch_size + 1)
@skipUnlessDBFeature("has_bulk_insert")
def test_max_batch_size(self):
objs = [Country(name=f"Country {i}") for i in range(1000)]
fields = ["name", "iso_two_letter", "description"]
max_batch_size = connection.ops.bulk_batch_size(fields, objs)
with self.assertNumQueries(ceil(len(objs) / max_batch_size)):
Country.objects.bulk_create(objs)
@skipUnlessDBFeature("has_bulk_insert") @skipUnlessDBFeature("has_bulk_insert")
def test_bulk_insert_expressions(self): def test_bulk_insert_expressions(self):
Restaurant.objects.bulk_create( Restaurant.objects.bulk_create(

11
tests/queries/test_bulk_update.py
View File

@@ -1,6 +1,8 @@
import datetime import datetime
from math import ceil
from django.core.exceptions import FieldDoesNotExist from django.core.exceptions import FieldDoesNotExist
from django.db import connection
from django.db.models import F from django.db.models import F
from django.db.models.functions import Lower from django.db.models.functions import Lower
from django.db.utils import IntegrityError from django.db.utils import IntegrityError
@@ -69,6 +71,15 @@ class BulkUpdateNoteTests(TestCase):
with self.assertNumQueries(len(self.notes)): with self.assertNumQueries(len(self.notes)):
Note.objects.bulk_update(self.notes, fields=["note"], batch_size=1) Note.objects.bulk_update(self.notes, fields=["note"], batch_size=1)
def test_max_batch_size(self):
max_batch_size = connection.ops.bulk_batch_size(
# PK is used twice, see comment in bulk_update().
[Note._meta.pk, Note._meta.pk, Note._meta.get_field("note")],
self.notes,
)
with self.assertNumQueries(ceil(len(self.notes) / max_batch_size)):
Note.objects.bulk_update(self.notes, fields=["note"])
def test_unsaved_models(self): def test_unsaved_models(self):
objs = self.notes + [Note(note="test", misc="test")] objs = self.notes + [Note(note="test", misc="test")]
msg = "All bulk_update() objects must have a primary key set." msg = "All bulk_update() objects must have a primary key set."