import copy
from django.contrib.contenttypes.fields import GenericForeignKey
from django.core import exceptions
from django.db.models import Manager
from django.db.models.constants import LOOKUP_SEP
from django.db.models.fields.related_descriptors import ForwardManyToOneDescriptor
from django.db.models.fields.related_descriptors import ForwardOneToOneDescriptor
from django.db.models.fields.related_descriptors import ManyToManyDescriptor
from django.db.models.fields.related_descriptors import ReverseManyToOneDescriptor
from django.db.models.fields.related_descriptors import ReverseOneToOneDescriptor
from django.db.models.query import normalize_prefetch_lookups
from django.db.models.query import prefetch_one_level
from django.utils.functional import cached_property
from django_prefetch_utils.selector import override_prefetch_related_objects
from .maps import PrefetchIdentityMap
from .wrappers import ForwardDescriptorPrefetchWrapper
from .wrappers import GenericForeignKeyPrefetchWrapper
from .wrappers import IdentityMapPrefetcher
from .wrappers import ManyToManyRelatedManagerWrapper
from .wrappers import ReverseManyToOneDescriptorPrefetchWrapper
from .wrappers import ReverseOneToOneDescriptorPrefetchWrapper
def get_identity_map_prefetcher(identity_map, descriptor, prefetcher):
if prefetcher is None:
return None
wrappers = {
ForwardManyToOneDescriptor: ForwardDescriptorPrefetchWrapper,
ForwardOneToOneDescriptor: ForwardDescriptorPrefetchWrapper,
ReverseOneToOneDescriptor: ReverseOneToOneDescriptorPrefetchWrapper,
ReverseManyToOneDescriptor: ReverseManyToOneDescriptorPrefetchWrapper,
ManyToManyDescriptor: ManyToManyRelatedManagerWrapper,
GenericForeignKey: GenericForeignKeyPrefetchWrapper,
}
wrapper_cls = wrappers.get(type(descriptor), IdentityMapPrefetcher)
return wrapper_cls(identity_map, prefetcher)
[docs]def get_prefetcher(obj_list, through_attr, to_attr):
"""
For the attribute *through_attr* on the given instance, finds
an object that has a ``get_prefetch_queryset()``.
Returns a 4 tuple containing:
- (the object with get_prefetch_queryset (or None),
- the descriptor object representing this relationship (or None),
- a boolean that is False if the attribute was not found at all,
- a list of the subset of *obj_list* that requires fetching
"""
instance = obj_list[0]
prefetcher = None
needs_fetching = obj_list
# For singly related objects, we have to avoid getting the attribute
# from the object, as this will trigger the query. So we first try
# on the class, in order to get the descriptor object.
rel_obj_descriptor = getattr(instance.__class__, through_attr, None)
if rel_obj_descriptor is None:
attr_found = hasattr(instance, through_attr)
else:
attr_found = True
# singly related object, descriptor object has the
# get_prefetch_queryset() method.
if hasattr(rel_obj_descriptor, "get_prefetch_queryset"):
prefetcher = rel_obj_descriptor
needs_fetching = [obj for obj in obj_list if not rel_obj_descriptor.is_cached(obj)]
else:
# descriptor doesn't support prefetching, so we go ahead and get
# the attribute on the instance rather than the class to
# support many related managers
rel_obj = getattr(instance, through_attr)
if hasattr(rel_obj, "get_prefetch_queryset"):
prefetcher = rel_obj
if through_attr != to_attr:
# Special case cached_property instances because hasattr
# triggers attribute computation and assignment.
if isinstance(getattr(instance.__class__, to_attr, None), cached_property):
needs_fetching = [obj for obj in obj_list if to_attr not in obj.__dict__]
else:
needs_fetching = [obj for obj in obj_list if not hasattr(obj, to_attr)]
else:
needs_fetching = [obj for obj in obj_list if through_attr not in obj._prefetched_objects_cache]
return prefetcher, rel_obj_descriptor, attr_found, needs_fetching
[docs]def get_prefetched_objects_from_list(obj_list, through_attr):
"""
Returns all of the related objects in *obj_list* from *through_attr*.
:type obj_list: list
:type through_attr: str
:rtype: list
"""
new_obj_list = []
for obj in obj_list:
if through_attr in getattr(obj, "_prefetched_objects_cache", ()):
# If related objects have been prefetched, use the
# cache rather than the object's through_attr.
new_obj = list(obj._prefetched_objects_cache.get(through_attr))
else:
try:
new_obj = getattr(obj, through_attr)
except exceptions.ObjectDoesNotExist:
continue
if new_obj is None:
continue
# We special-case `list` rather than something more generic
# like `Iterable` because we don't want to accidentally match
# user models that define __iter__.
if isinstance(new_obj, list):
new_obj_list.extend(new_obj)
elif isinstance(new_obj, Manager):
# This case in needed for Django < 2.1 where the RelatedManager
# returns the wrong cache name so that *through_attr* does not
# appear in _prefetched_objects_cache. See Django #28723.
new_obj_list.extend(new_obj.all())
else:
new_obj_list.append(new_obj)
return new_obj_list
[docs]def get_default_prefetch_identity_map():
"""
Returns an empty default identity map for use during prefetching.
:rtype: :class:`django_prefetch_utils.identity_map.maps.PrefetchIdentityMap`
"""
return PrefetchIdentityMap()
[docs]def use_prefetch_identity_map():
"""
A context decorator which enables the identity map version of
``prefetch_related_objects``::
with use_prefetch_identity_map():
dogs = list(Dogs.objects.prefetch_related('toys'))
.. note::
A new identity map is created and used for each call of
``prefetched_related_objects``.
"""
return override_prefetch_related_objects(prefetch_related_objects)