Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/django/db/models/manager.py: 52%
101 statements
« prev ^ index » next coverage.py v7.0.5, created at 2023-01-17 06:13 +0000
1import copy
2import inspect
3from functools import wraps
4from importlib import import_module
6from django.db import router
7from django.db.models.query import QuerySet
10class BaseManager:
11 # To retain order, track each time a Manager instance is created.
12 creation_counter = 0
14 # Set to True for the 'objects' managers that are automatically created.
15 auto_created = False
17 #: If set to True the manager will be serialized into migrations and will
18 #: thus be available in e.g. RunPython operations.
19 use_in_migrations = False
21 def __new__(cls, *args, **kwargs):
22 # Capture the arguments to make returning them trivial.
23 obj = super().__new__(cls)
24 obj._constructor_args = (args, kwargs)
25 return obj
27 def __init__(self):
28 super().__init__()
29 self._set_creation_counter()
30 self.model = None
31 self.name = None
32 self._db = None
33 self._hints = {}
35 def __str__(self):
36 """Return "app_label.model_label.manager_name"."""
37 return "%s.%s" % (self.model._meta.label, self.name)
39 def __class_getitem__(cls, *args, **kwargs):
40 return cls
42 def deconstruct(self):
43 """
44 Return a 5-tuple of the form (as_manager (True), manager_class,
45 queryset_class, args, kwargs).
47 Raise a ValueError if the manager is dynamically generated.
48 """
49 qs_class = self._queryset_class
50 if getattr(self, "_built_with_as_manager", False):
51 # using MyQuerySet.as_manager()
52 return (
53 True, # as_manager
54 None, # manager_class
55 "%s.%s" % (qs_class.__module__, qs_class.__name__), # qs_class
56 None, # args
57 None, # kwargs
58 )
59 else:
60 module_name = self.__module__
61 name = self.__class__.__name__
62 # Make sure it's actually there and not an inner class
63 module = import_module(module_name)
64 if not hasattr(module, name):
65 raise ValueError(
66 "Could not find manager %s in %s.\n"
67 "Please note that you need to inherit from managers you "
68 "dynamically generated with 'from_queryset()'."
69 % (name, module_name)
70 )
71 return (
72 False, # as_manager
73 "%s.%s" % (module_name, name), # manager_class
74 None, # qs_class
75 self._constructor_args[0], # args
76 self._constructor_args[1], # kwargs
77 )
79 def check(self, **kwargs):
80 return []
82 @classmethod
83 def _get_queryset_methods(cls, queryset_class):
84 def create_method(name, method):
85 @wraps(method)
86 def manager_method(self, *args, **kwargs):
87 return getattr(self.get_queryset(), name)(*args, **kwargs)
89 return manager_method
91 new_methods = {}
92 for name, method in inspect.getmembers(
93 queryset_class, predicate=inspect.isfunction
94 ):
95 # Only copy missing methods.
96 if hasattr(cls, name):
97 continue
98 # Only copy public methods or methods with the attribute
99 # queryset_only=False.
100 queryset_only = getattr(method, "queryset_only", None)
101 if queryset_only or (queryset_only is None and name.startswith("_")):
102 continue
103 # Copy the method onto the manager.
104 new_methods[name] = create_method(name, method)
105 return new_methods
107 @classmethod
108 def from_queryset(cls, queryset_class, class_name=None):
109 if class_name is None:
110 class_name = "%sFrom%s" % (cls.__name__, queryset_class.__name__)
111 return type(
112 class_name,
113 (cls,),
114 {
115 "_queryset_class": queryset_class,
116 **cls._get_queryset_methods(queryset_class),
117 },
118 )
120 def contribute_to_class(self, cls, name):
121 self.name = self.name or name
122 self.model = cls
124 setattr(cls, name, ManagerDescriptor(self))
126 cls._meta.add_manager(self)
128 def _set_creation_counter(self):
129 """
130 Set the creation counter value for this instance and increment the
131 class-level copy.
132 """
133 self.creation_counter = BaseManager.creation_counter
134 BaseManager.creation_counter += 1
136 def db_manager(self, using=None, hints=None):
137 obj = copy.copy(self)
138 obj._db = using or self._db
139 obj._hints = hints or self._hints
140 return obj
142 @property
143 def db(self):
144 return self._db or router.db_for_read(self.model, **self._hints)
146 #######################
147 # PROXIES TO QUERYSET #
148 #######################
150 def get_queryset(self):
151 """
152 Return a new QuerySet object. Subclasses can override this method to
153 customize the behavior of the Manager.
154 """
155 return self._queryset_class(model=self.model, using=self._db, hints=self._hints)
157 def all(self):
158 # We can't proxy this method through the `QuerySet` like we do for the
159 # rest of the `QuerySet` methods. This is because `QuerySet.all()`
160 # works by creating a "copy" of the current queryset and in making said
161 # copy, all the cached `prefetch_related` lookups are lost. See the
162 # implementation of `RelatedManager.get_queryset()` for a better
163 # understanding of how this comes into play.
164 return self.get_queryset()
166 def __eq__(self, other):
167 return (
168 isinstance(other, self.__class__)
169 and self._constructor_args == other._constructor_args
170 )
172 def __hash__(self):
173 return id(self)
176class Manager(BaseManager.from_queryset(QuerySet)):
177 pass
180class ManagerDescriptor:
181 def __init__(self, manager):
182 self.manager = manager
184 def __get__(self, instance, cls=None):
185 if instance is not None:
186 raise AttributeError(
187 "Manager isn't accessible via %s instances" % cls.__name__
188 )
190 if cls._meta.abstract:
191 raise AttributeError(
192 "Manager isn't available; %s is abstract" % (cls._meta.object_name,)
193 )
195 if cls._meta.swapped:
196 raise AttributeError(
197 "Manager isn't available; '%s' has been swapped for '%s'"
198 % (
199 cls._meta.label,
200 cls._meta.swapped,
201 )
202 )
204 return cls._meta.managers_map[self.manager.name]
207class EmptyManager(Manager):
208 def __init__(self, model):
209 super().__init__()
210 self.model = model
212 def get_queryset(self):
213 return super().get_queryset().none()