1 from __future__
import with_statement
, absolute_import
3 from operator
import attrgetter
5 from django
.db
import connection
6 from django
.test
import TestCase
, skipIfDBFeature
7 from django
.test
.utils
import override_settings
9 from .models
import Country
, Restaurant
, Pizzeria
, State
, TwoFields
12 class BulkCreateTests(TestCase
):
15 Country(name
="United States of America", iso_two_letter
="US"),
16 Country(name
="The Netherlands", iso_two_letter
="NL"),
17 Country(name
="Germany", iso_two_letter
="DE"),
18 Country(name
="Czech Republic", iso_two_letter
="CZ")
21 def test_simple(self
):
22 created
= Country
.objects
.bulk_create(self
.data
)
23 self
.assertEqual(len(created
), 4)
24 self
.assertQuerysetEqual(Country
.objects
.order_by("-name"), [
25 "United States of America", "The Netherlands", "Germany", "Czech Republic"
26 ], attrgetter("name"))
28 created
= Country
.objects
.bulk_create([])
29 self
.assertEqual(created
, [])
30 self
.assertEqual(Country
.objects
.count(), 4)
32 def test_efficiency(self
):
33 with self
.assertNumQueries(1):
34 Country
.objects
.bulk_create(self
.data
)
36 def test_inheritance(self
):
37 Restaurant
.objects
.bulk_create([
38 Restaurant(name
="Nicholas's")
40 self
.assertQuerysetEqual(Restaurant
.objects
.all(), [
42 ], attrgetter("name"))
43 with self
.assertRaises(ValueError):
44 Pizzeria
.objects
.bulk_create([
45 Pizzeria(name
="The Art of Pizza")
47 self
.assertQuerysetEqual(Pizzeria
.objects
.all(), [])
48 self
.assertQuerysetEqual(Restaurant
.objects
.all(), [
50 ], attrgetter("name"))
52 def test_non_auto_increment_pk(self
):
53 with self
.assertNumQueries(1):
54 State
.objects
.bulk_create([
55 State(two_letter_code
=s
)
56 for s
in ["IL", "NY", "CA", "ME"]
58 self
.assertQuerysetEqual(State
.objects
.order_by("two_letter_code"), [
59 "CA", "IL", "ME", "NY",
60 ], attrgetter("two_letter_code"))
62 def test_large_batch(self
):
63 with
override_settings(DEBUG
=True):
64 connection
.queries
= []
65 TwoFields
.objects
.bulk_create([
66 TwoFields(f1
=i
, f2
=i
+1) for i
in range(0, 1001)
68 self
.assertTrue(len(connection
.queries
) < 10)
69 self
.assertEqual(TwoFields
.objects
.count(), 1001)
71 TwoFields
.objects
.filter(f1__gte
=450, f1__lte
=550).count(),
73 self
.assertEqual(TwoFields
.objects
.filter(f2__gte
=901).count(), 101)
75 def test_large_batch_mixed(self
):
77 Test inserting a large batch with objects having primary key set
78 mixed together with objects without PK set.
80 with
override_settings(DEBUG
=True):
81 connection
.queries
= []
82 TwoFields
.objects
.bulk_create([
83 TwoFields(id=i
if i
% 2 == 0 else None, f1
=i
, f2
=i
+1)
84 for i
in range(100000, 101000)])
85 self
.assertTrue(len(connection
.queries
) < 10)
86 self
.assertEqual(TwoFields
.objects
.count(), 1000)
87 # We can't assume much about the ID's created, except that the above
88 # created IDs must exist.
89 id_range
= range(100000, 101000, 2)
90 self
.assertEqual(TwoFields
.objects
.filter(id__in
=id_range
).count(), 500)
91 self
.assertEqual(TwoFields
.objects
.exclude(id__in
=id_range
).count(), 500)
93 def test_explicit_batch_size(self
):
94 objs
= [TwoFields(f1
=i
, f2
=i
) for i
in range(0, 100)]
95 with self
.assertNumQueries(2):
96 TwoFields
.objects
.bulk_create(objs
, 50)
97 TwoFields
.objects
.all().delete()
98 with self
.assertNumQueries(1):
99 TwoFields
.objects
.bulk_create(objs
, len(objs
))