10 examples of 'django prefetch_related' in Python

53
def test_m2m_forward(self):
54
    with self.assertNumQueries(2):
55
        lists = [list(b.authors.all()) for b in Book.objects.prefetch_related('authors')]
56
57
    normal_lists = [list(b.authors.all()) for b in Book.objects.all()]
58
    self.assertEqual(lists, normal_lists)

706
@property
707
def prefetch_related(self):
708
    """
709
    With main and attached records, prefetch_related lists can be
710
    generated by combining the select_related lists for attached
711
    children and prefetch_related lists for all children.
712
    """
713
    try:
714
        self._prefetch_related = self._prefetch_related
715
    except AttributeError:
716
        att_sr = self.combine_rels_from_children('select_related',
717
                                                 self.attached_children)
718
        all_pr = self.combine_rels_from_children('prefetch_related')
719
        self._prefetch_related = self.combine_lists(all_pr, att_sr)
720
    return self._prefetch_related

71
def get_prefetch_queryset(self, instances, queryset=None):
72
    if queryset is not None:
73
        raise ValueError("Custom queryset can't be used for this lookup.")
74
75
    instance = instances[0]
76
    db = self._db or router.db_for_read(type(instance), instance=instance)
77
78
    fieldname = (
79
        "object_id"
80
        if issubclass(self.through, CommonGenericTaggedItemBase)
81
        else "content_object"
82
    )
83
    fk = self.through._meta.get_field(fieldname)
84
    query = {
85
        "%s__%s__in"
86
        % (self.through.tag_relname(), fk.name): {
87
            obj._get_pk_val() for obj in instances
88
        }
89
    }
90
    join_table = self.through._meta.db_table
91
    source_col = fk.column
92
    connection = connections[db]
93
    qn = connection.ops.quote_name
94
    qs = (
95
        self.get_queryset(query)
96
        .using(db)
97
        .extra(
98
            select={
99
                "_prefetch_related_val": "{}.{}".format(
100
                    qn(join_table), qn(source_col)
101
                )
102
            }
103
        )
104
    )
105
    if VERSION < (2, 0):
106
        return (
107
            qs,
108
            attrgetter("_prefetch_related_val"),
109
            lambda obj: obj._get_pk_val(),
110
            False,
111
            self.prefetch_cache_name,
112
        )
113
    else:
114
        return (
115
            qs,
116
            attrgetter("_prefetch_related_val"),
117
            lambda obj: obj._get_pk_val(),
118
            False,
119
            self.prefetch_cache_name,
120
            False,
121
        )

249
def test_it_prefetches_foreign_key_relations_from_reverse_side_with_serializer_that_has_source(self):
250
    # Need two queries because of prefetch related
251
    data = _run_test(TopLevelSerializerWithNestedSerializerWithSource, TopLevel, sql_queries=2)
252
253
    assert len(data) == 1
254
    assert len(data[0]['kiddos']) == 3

20
async def test_prefetch_object(self):
21
    tournament = await Tournament.create(name="tournament")
22
    await Event.create(name="First", tournament=tournament)
23
    await Event.create(name="Second", tournament=tournament)
24
    tournament_with_filtered = (
25
        await Tournament.all()
26
        .prefetch_related(Prefetch("events", queryset=Event.filter(name="First")))
27
        .first()
28
    )
29
    tournament = await Tournament.first().prefetch_related("events")
30
    self.assertEqual(len(tournament_with_filtered.events), 1)
31
    self.assertEqual(len(tournament.events), 2)

24
def get_query_set(self, *args, **kwargs):
25
    query = super(PartManager, self).get_query_set(*args, **kwargs)
26
    query = query.select_related('designation__description',
27
                                 'supplier')
28
29
    query = query.prefetch_related('analogs', 'images')
30
    return query

149
async def test_get_m2m_forward_prefetch_related(self):
150
    obj1 = await self.model.create(chars="aaa")
151
    obj2 = await self.model.create(chars="bbb")
152
    await obj1.rel_to.add(obj2)
153
154
    obj2a = await self.model.get(eyedee=obj2.eyedee).prefetch_related("rel_from")
155
    self.assertEqual(list(obj2a.rel_from), [obj1])

244
def prefetch_export_runs(queryset_list_or_model):
245
    prefetch_args = ['job__provider_tasks__provider', 'job__provider_tasks__formats',
246
                     'provider_tasks__tasks__result', 'provider_tasks__tasks__exceptions']
247
    if isinstance(queryset_list_or_model, models.query.QuerySet):
248
        return queryset_list_or_model.select_related('user').prefetch_related(*prefetch_args)
249
    elif isinstance(queryset_list_or_model, list):
250
        models.prefetch_related_objects(queryset_list_or_model, *prefetch_args)
251
    elif isinstance(queryset_list_or_model, ExportRun):
252
        models.prefetch_related_objects([queryset_list_or_model], *prefetch_args)
253
    return queryset_list_or_model

235
async def test_prefetch_related_missing_field(self):
236
    tournament = await Tournament.create(name="New Tournament")
237
    await Event.create(name="Test", tournament_id=tournament.id)
238
239
    with self.assertRaisesRegex(FieldError, "Relation tourn1ment for models.Event not found"):
240
        await Event.filter(name="Test").prefetch_related("tourn1ment")

586
def test_remove_after_prefetch(self):
587
    a4 = Article.objects.prefetch_related('publications').get(id=self.a4.id)
588
    self.assertQuerysetEqual(a4.publications.all(), [''])
589
    a4.publications.remove(self.p2)
590
    self.assertQuerysetEqual(a4.publications.all(), [])