Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/apps/registry.py: 77%
176 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1import functools
2import sys
3import threading
4import warnings
5from collections import Counter, defaultdict
6from functools import partial
8from django.core.exceptions import AppRegistryNotReady, ImproperlyConfigured
10from .config import AppConfig
13class Apps:
14 """
15 A registry that stores the configuration of installed applications.
17 It also keeps track of models, e.g. to provide reverse relations.
18 """
20 def __init__(self, installed_apps=()):
21 # installed_apps is set to None when creating the master registry
22 # because it cannot be populated at that point. Other registries must
23 # provide a list of installed apps and are populated immediately.
24 if installed_apps is None and hasattr(sys.modules[__name__], "apps"): 24 ↛ 25line 24 didn't jump to line 25, because the condition on line 24 was never true
25 raise RuntimeError("You must supply an installed_apps argument.")
27 # Mapping of app labels => model names => model classes. Every time a
28 # model is imported, ModelBase.__new__ calls apps.register_model which
29 # creates an entry in all_models. All imported models are registered,
30 # regardless of whether they're defined in an installed application
31 # and whether the registry has been populated. Since it isn't possible
32 # to reimport a module safely (it could reexecute initialization code)
33 # all_models is never overridden or reset.
34 self.all_models = defaultdict(dict)
36 # Mapping of labels to AppConfig instances for installed apps.
37 self.app_configs = {}
39 # Stack of app_configs. Used to store the current state in
40 # set_available_apps and set_installed_apps.
41 self.stored_app_configs = []
43 # Whether the registry is populated.
44 self.apps_ready = self.models_ready = self.ready = False
45 # For the autoreloader.
46 self.ready_event = threading.Event()
48 # Lock for thread-safe population.
49 self._lock = threading.RLock()
50 self.loading = False
52 # Maps ("app_label", "modelname") tuples to lists of functions to be
53 # called when the corresponding model is ready. Used by this class's
54 # `lazy_model_operation()` and `do_pending_operations()` methods.
55 self._pending_operations = defaultdict(list)
57 # Populate apps and models, unless it's the master registry.
58 if installed_apps is not None:
59 self.populate(installed_apps)
61 def populate(self, installed_apps=None):
62 """
63 Load application configurations and models.
65 Import each application module and then each model module.
67 It is thread-safe and idempotent, but not reentrant.
68 """
69 if self.ready: 69 ↛ 70line 69 didn't jump to line 70, because the condition on line 69 was never true
70 return
72 # populate() might be called by two threads in parallel on servers
73 # that create threads before initializing the WSGI callable.
74 with self._lock:
75 if self.ready: 75 ↛ 76line 75 didn't jump to line 76, because the condition on line 75 was never true
76 return
78 # An RLock prevents other threads from entering this section. The
79 # compare and set operation below is atomic.
80 if self.loading: 80 ↛ 83line 80 didn't jump to line 83, because the condition on line 80 was never true
81 # Prevent reentrant calls to avoid running AppConfig.ready()
82 # methods twice.
83 raise RuntimeError("populate() isn't reentrant")
84 self.loading = True
86 # Phase 1: initialize app configs and import app modules.
87 for entry in installed_apps:
88 if isinstance(entry, AppConfig):
89 app_config = entry
90 else:
91 app_config = AppConfig.create(entry)
92 if app_config.label in self.app_configs: 92 ↛ 93line 92 didn't jump to line 93, because the condition on line 92 was never true
93 raise ImproperlyConfigured(
94 "Application labels aren't unique, "
95 "duplicates: %s" % app_config.label
96 )
98 self.app_configs[app_config.label] = app_config
99 app_config.apps = self
101 # Check for duplicate app names.
102 counts = Counter(
103 app_config.name for app_config in self.app_configs.values()
104 )
105 duplicates = [name for name, count in counts.most_common() if count > 1]
106 if duplicates: 106 ↛ 107line 106 didn't jump to line 107, because the condition on line 106 was never true
107 raise ImproperlyConfigured(
108 "Application names aren't unique, "
109 "duplicates: %s" % ", ".join(duplicates)
110 )
112 self.apps_ready = True
114 # Phase 2: import models modules.
115 for app_config in self.app_configs.values():
116 app_config.import_models()
118 self.clear_cache()
120 self.models_ready = True
122 # Phase 3: run ready() methods of app configs.
123 for app_config in self.get_app_configs():
124 app_config.ready()
126 self.ready = True
127 self.ready_event.set()
129 def check_apps_ready(self):
130 """Raise an exception if all apps haven't been imported yet."""
131 if not self.apps_ready: 131 ↛ 132line 131 didn't jump to line 132, because the condition on line 131 was never true
132 from django.conf import settings
134 # If "not ready" is due to unconfigured settings, accessing
135 # INSTALLED_APPS raises a more helpful ImproperlyConfigured
136 # exception.
137 settings.INSTALLED_APPS
138 raise AppRegistryNotReady("Apps aren't loaded yet.")
140 def check_models_ready(self):
141 """Raise an exception if all models haven't been imported yet."""
142 if not self.models_ready: 142 ↛ 143line 142 didn't jump to line 143, because the condition on line 142 was never true
143 raise AppRegistryNotReady("Models aren't loaded yet.")
145 def get_app_configs(self):
146 """Import applications and return an iterable of app configs."""
147 self.check_apps_ready()
148 return self.app_configs.values()
150 def get_app_config(self, app_label):
151 """
152 Import applications and returns an app config for the given label.
154 Raise LookupError if no application exists with this label.
155 """
156 self.check_apps_ready()
157 try:
158 return self.app_configs[app_label]
159 except KeyError:
160 message = "No installed app with label '%s'." % app_label
161 for app_config in self.get_app_configs():
162 if app_config.name == app_label: 162 ↛ 163line 162 didn't jump to line 163, because the condition on line 162 was never true
163 message += " Did you mean '%s'?" % app_config.label
164 break
165 raise LookupError(message)
167 # This method is performance-critical at least for Django's test suite.
168 @functools.lru_cache(maxsize=None)
169 def get_models(self, include_auto_created=False, include_swapped=False):
170 """
171 Return a list of all installed models.
173 By default, the following models aren't included:
175 - auto-created models for many-to-many relations without
176 an explicit intermediate table,
177 - models that have been swapped out.
179 Set the corresponding keyword argument to True to include such models.
180 """
181 self.check_models_ready()
183 result = []
184 for app_config in self.app_configs.values():
185 result.extend(app_config.get_models(include_auto_created, include_swapped))
186 return result
188 def get_model(self, app_label, model_name=None, require_ready=True):
189 """
190 Return the model matching the given app_label and model_name.
192 As a shortcut, app_label may be in the form <app_label>.<model_name>.
194 model_name is case-insensitive.
196 Raise LookupError if no application exists with this label, or no
197 model exists with this name in the application. Raise ValueError if
198 called with a single argument that doesn't contain exactly one dot.
199 """
200 if require_ready:
201 self.check_models_ready()
202 else:
203 self.check_apps_ready()
205 if model_name is None:
206 app_label, model_name = app_label.split(".")
208 app_config = self.get_app_config(app_label)
210 if not require_ready and app_config.models is None: 210 ↛ 211line 210 didn't jump to line 211, because the condition on line 210 was never true
211 app_config.import_models()
213 return app_config.get_model(model_name, require_ready=require_ready)
215 def register_model(self, app_label, model):
216 # Since this method is called when models are imported, it cannot
217 # perform imports because of the risk of import loops. It mustn't
218 # call get_app_config().
219 model_name = model._meta.model_name
220 app_models = self.all_models[app_label]
221 if model_name in app_models: 221 ↛ 222line 221 didn't jump to line 222, because the condition on line 221 was never true
222 if (
223 model.__name__ == app_models[model_name].__name__
224 and model.__module__ == app_models[model_name].__module__
225 ):
226 warnings.warn(
227 "Model '%s.%s' was already registered. Reloading models is not "
228 "advised as it can lead to inconsistencies, most notably with "
229 "related models." % (app_label, model_name),
230 RuntimeWarning,
231 stacklevel=2,
232 )
233 else:
234 raise RuntimeError(
235 "Conflicting '%s' models in application '%s': %s and %s."
236 % (model_name, app_label, app_models[model_name], model)
237 )
238 app_models[model_name] = model
239 self.do_pending_operations(model)
240 self.clear_cache()
242 def is_installed(self, app_name):
243 """
244 Check whether an application with this name exists in the registry.
246 app_name is the full name of the app e.g. 'django.contrib.admin'.
247 """
248 self.check_apps_ready()
249 return any(ac.name == app_name for ac in self.app_configs.values()) 249 ↛ exitline 249 didn't finish the generator expression on line 249
251 def get_containing_app_config(self, object_name):
252 """
253 Look for an app config containing a given object.
255 object_name is the dotted Python path to the object.
257 Return the app config for the inner application in case of nesting.
258 Return None if the object isn't in any registered app config.
259 """
260 self.check_apps_ready()
261 candidates = []
262 for app_config in self.app_configs.values():
263 if object_name.startswith(app_config.name):
264 subpath = object_name[len(app_config.name) :]
265 if subpath == "" or subpath[0] == ".":
266 candidates.append(app_config)
267 if candidates:
268 return sorted(candidates, key=lambda ac: -len(ac.name))[0]
270 def get_registered_model(self, app_label, model_name):
271 """
272 Similar to get_model(), but doesn't require that an app exists with
273 the given app_label.
275 It's safe to call this method at import time, even while the registry
276 is being populated.
277 """
278 model = self.all_models[app_label].get(model_name.lower())
279 if model is None:
280 raise LookupError("Model '%s.%s' not registered." % (app_label, model_name))
281 return model
283 @functools.lru_cache(maxsize=None)
284 def get_swappable_settings_name(self, to_string):
285 """
286 For a given model string (e.g. "auth.User"), return the name of the
287 corresponding settings name if it refers to a swappable model. If the
288 referred model is not swappable, return None.
290 This method is decorated with lru_cache because it's performance
291 critical when it comes to migrations. Since the swappable settings don't
292 change after Django has loaded the settings, there is no reason to get
293 the respective settings attribute over and over again.
294 """
295 to_string = to_string.lower()
296 for model in self.get_models(include_swapped=True):
297 swapped = model._meta.swapped
298 # Is this model swapped out for the model given by to_string?
299 if swapped and swapped.lower() == to_string:
300 return model._meta.swappable
301 # Is this model swappable and the one given by to_string?
302 if model._meta.swappable and model._meta.label_lower == to_string: 302 ↛ 303line 302 didn't jump to line 303, because the condition on line 302 was never true
303 return model._meta.swappable
304 return None
306 def set_available_apps(self, available):
307 """
308 Restrict the set of installed apps used by get_app_config[s].
310 available must be an iterable of application names.
312 set_available_apps() must be balanced with unset_available_apps().
314 Primarily used for performance optimization in TransactionTestCase.
316 This method is safe in the sense that it doesn't trigger any imports.
317 """
318 available = set(available)
319 installed = {app_config.name for app_config in self.get_app_configs()}
320 if not available.issubset(installed):
321 raise ValueError(
322 "Available apps isn't a subset of installed apps, extra apps: %s"
323 % ", ".join(available - installed)
324 )
326 self.stored_app_configs.append(self.app_configs)
327 self.app_configs = {
328 label: app_config
329 for label, app_config in self.app_configs.items()
330 if app_config.name in available
331 }
332 self.clear_cache()
334 def unset_available_apps(self):
335 """Cancel a previous call to set_available_apps()."""
336 self.app_configs = self.stored_app_configs.pop()
337 self.clear_cache()
339 def set_installed_apps(self, installed):
340 """
341 Enable a different set of installed apps for get_app_config[s].
343 installed must be an iterable in the same format as INSTALLED_APPS.
345 set_installed_apps() must be balanced with unset_installed_apps(),
346 even if it exits with an exception.
348 Primarily used as a receiver of the setting_changed signal in tests.
350 This method may trigger new imports, which may add new models to the
351 registry of all imported models. They will stay in the registry even
352 after unset_installed_apps(). Since it isn't possible to replay
353 imports safely (e.g. that could lead to registering listeners twice),
354 models are registered when they're imported and never removed.
355 """
356 if not self.ready:
357 raise AppRegistryNotReady("App registry isn't ready yet.")
358 self.stored_app_configs.append(self.app_configs)
359 self.app_configs = {}
360 self.apps_ready = self.models_ready = self.loading = self.ready = False
361 self.clear_cache()
362 self.populate(installed)
364 def unset_installed_apps(self):
365 """Cancel a previous call to set_installed_apps()."""
366 self.app_configs = self.stored_app_configs.pop()
367 self.apps_ready = self.models_ready = self.ready = True
368 self.clear_cache()
370 def clear_cache(self):
371 """
372 Clear all internal caches, for methods that alter the app registry.
374 This is mostly used in tests.
375 """
376 # Call expire cache on each model. This will purge
377 # the relation tree and the fields cache.
378 self.get_models.cache_clear()
379 if self.ready:
380 # Circumvent self.get_models() to prevent that the cache is refilled.
381 # This particularly prevents that an empty value is cached while cloning.
382 for app_config in self.app_configs.values():
383 for model in app_config.get_models(include_auto_created=True):
384 model._meta._expire_cache()
386 def lazy_model_operation(self, function, *model_keys):
387 """
388 Take a function and a number of ("app_label", "modelname") tuples, and
389 when all the corresponding models have been imported and registered,
390 call the function with the model classes as its arguments.
392 The function passed to this method must accept exactly n models as
393 arguments, where n=len(model_keys).
394 """
395 # Base case: no arguments, just execute the function.
396 if not model_keys:
397 function()
398 # Recursive case: take the head of model_keys, wait for the
399 # corresponding model class to be imported and registered, then apply
400 # that argument to the supplied function. Pass the resulting partial
401 # to lazy_model_operation() along with the remaining model args and
402 # repeat until all models are loaded and all arguments are applied.
403 else:
404 next_model, *more_models = model_keys
406 # This will be executed after the class corresponding to next_model
407 # has been imported and registered. The `func` attribute provides
408 # duck-type compatibility with partials.
409 def apply_next_model(model):
410 next_function = partial(apply_next_model.func, model)
411 self.lazy_model_operation(next_function, *more_models)
413 apply_next_model.func = function
415 # If the model has already been imported and registered, partially
416 # apply it to the function now. If not, add it to the list of
417 # pending operations for the model, where it will be executed with
418 # the model class as its sole argument once the model is ready.
419 try:
420 model_class = self.get_registered_model(*next_model)
421 except LookupError:
422 self._pending_operations[next_model].append(apply_next_model)
423 else:
424 apply_next_model(model_class)
426 def do_pending_operations(self, model):
427 """
428 Take a newly-prepared model and pass it to each function waiting for
429 it. This is called at the very end of Apps.register_model().
430 """
431 key = model._meta.app_label, model._meta.model_name
432 for function in self._pending_operations.pop(key, []):
433 function(model)
436apps = Apps(installed_apps=None)