Every line of 'django prefetch_related' code snippets is scanned for vulnerabilities by our powerful machine learning engine that combs millions of open source libraries, ensuring your Python code is secure.
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(), [])