Coverage for src/backend/InvenTree/InvenTree/ready.py: 75%
105 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Functions to check if certain parts of InvenTree are ready."""
3import functools
4import inspect
5import os
6import sys
7import warnings
9from django.conf import settings
11import structlog
13logger = structlog.get_logger('inventree')
16# Keep track of loaded apps, to prevent multiple executions of ready functions
17_loaded_apps = set()
20def clearLoadedApps():
21 """Clear the set of loaded apps."""
22 global _loaded_apps
23 _loaded_apps = set()
26def setAppLoaded(app_name: str):
27 """Mark an app as loaded."""
28 global _loaded_apps
29 _loaded_apps.add(app_name)
32def isAppLoaded(app_name: str) -> bool:
33 """Return True if the app has been marked as loaded."""
34 global _loaded_apps
35 return app_name in _loaded_apps
38def isInTestMode():
39 """Returns True if the database is in testing mode."""
40 return any(x in sys.argv for x in ['test', 'pytest']) or sys.argv[0].endswith(
41 'pytest'
42 )
45def isWaitingForDatabase():
46 """Return True if we are currently waiting for the database to be ready."""
47 return 'wait_for_db' in sys.argv
50def isImportingData():
51 """Returns True if the database is currently importing (or exporting) data, e.g. 'loaddata' command is performed."""
52 return any(x in sys.argv for x in ['flush', 'loaddata', 'dumpdata'])
55def isRunningMigrations():
56 """Return True if the database is currently running migrations."""
57 return any(
58 x in sys.argv
59 for x in ['migrate', 'makemigrations', 'showmigrations', 'runmigrations']
60 )
63def isRebuildingData():
64 """Return true if any of the rebuilding commands are being executed."""
65 return any(
66 x in sys.argv
67 for x in [
68 'rebuild',
69 'rebuild_models',
70 'rebuild_thumbnails',
71 'remove_stale_contenttypes',
72 ]
73 )
76def isRunningBackup():
77 """Return true if any of the backup commands are being executed."""
78 return any(
79 x in sys.argv
80 for x in [
81 'backup',
82 'restore',
83 'dbbackup',
84 'dbrestore',
85 'mediabackup',
86 'mediarestore',
87 ]
88 )
91def isCollectingPlugins():
92 """Return True if the 'collectplugins' command is being executed."""
93 return 'collectplugins' in sys.argv
96# This variable is used to cache the result of the isGeneratingSchema function, to prevent multiple executions of the same checks
97_IS_GENERATING_SCHEMA: bool | None = None
100def _setGeneratingSchema(value: bool):
101 """Set the value of the isGeneratingSchema variable."""
102 global _IS_GENERATING_SCHEMA
103 _IS_GENERATING_SCHEMA = value
104 return value
107def isGeneratingSchema():
108 """Return true if schema generation is being executed."""
109 global _IS_GENERATING_SCHEMA
111 if _IS_GENERATING_SCHEMA is not None:
112 return _IS_GENERATING_SCHEMA
114 if isInServerThread() or isInWorkerThread(): 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 return _setGeneratingSchema(False)
117 if isRunningMigrations() or isRunningBackup() or isRebuildingData(): 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true
118 return _setGeneratingSchema(False)
120 if isImportingData(): 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true
121 return _setGeneratingSchema(False)
123 if isInTestMode(): 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true
124 return _setGeneratingSchema(False)
126 if isWaitingForDatabase(): 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 return _setGeneratingSchema(False)
129 if isCollectingPlugins(): 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true
130 return _setGeneratingSchema(False)
132 # Additional set of commands which should not trigger schema generation
133 excluded_commands = [
134 'compilemessages',
135 'createsuperuser',
136 'clean_settings',
137 'collectstatic',
138 'makemessages',
139 'wait_for_db',
140 'list_apps',
141 'gunicorn',
142 'sqlflush',
143 'qcluster',
144 'check',
145 'shell',
146 'help',
147 ]
149 if any(cmd in sys.argv for cmd in excluded_commands): 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true
150 return _setGeneratingSchema(False)
152 included_commands = [
153 'schema',
154 'spectactular',
155 # schema adjacent calls
156 'export_settings_definitions',
157 'export_tags',
158 'export_filters',
159 'export_report_context',
160 ]
162 if any(cmd in sys.argv for cmd in included_commands): 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true
163 return _setGeneratingSchema(True)
165 # This is a very inefficient call - so we only use it as a last resort
166 result = any('drf_spectacular' in frame.filename for frame in inspect.stack())
168 if not result: 168 ↛ 178line 168 didn't jump to line 178 because the condition on line 168 was always true
169 # We should only get here if we *are* generating schema
170 # Raise a warning, so that developers can add extra checks above
172 if settings.DEBUG: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true
173 logger.warning(
174 'isGeneratingSchema called outside of expected contexts - this may be a sign of a problem with the ready() function'
175 )
176 logger.warning('sys.argv: %s', sys.argv)
178 return _setGeneratingSchema(result)
181def isInWorkerThread():
182 """Returns True if the current thread is a background worker thread."""
183 return 'qcluster' in sys.argv
186def isInServerThread():
187 """Returns True if the current thread is a server thread."""
188 if isInWorkerThread(): 188 ↛ 189line 188 didn't jump to line 189 because the condition on line 188 was never true
189 return False
191 if 'runserver' in sys.argv: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 return True
194 return 'gunicorn' in sys.argv[0]
197def isInMainThread():
198 """Django runserver starts two processes, one for the actual dev server and the other to reload the application.
200 - The RUN_MAIN env is set in that case. However if --noreload is applied, this variable
201 is not set because there are no different threads.
202 """
203 if 'runserver' in sys.argv and '--noreload' not in sys.argv: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true
204 return os.environ.get('RUN_MAIN', None) == 'true'
206 return not isInWorkerThread()
209def readOnlyCommands():
210 """Return a list of read-only management commands which should not trigger database writes."""
211 return [
212 'help',
213 'check',
214 'shell',
215 'sqlflush',
216 'list_apps',
217 'wait_for_db',
218 'spectactular',
219 'makemessages',
220 'collectstatic',
221 'showmigrations',
222 'compilemessages',
223 ]
226def isReadOnlyCommand():
227 """Return True if the current command is a read-only command, which should not trigger any database writes."""
228 if ( 228 ↛ 234line 228 didn't jump to line 234 because the condition on line 228 was never true
229 isImportingData()
230 or isRunningMigrations()
231 or isRebuildingData()
232 or isRunningBackup()
233 ):
234 return True
236 return any(cmd in sys.argv for cmd in readOnlyCommands())
239def canAppAccessDatabase(
240 allow_test: bool = False, allow_plugins: bool = False, allow_shell: bool = False
241):
242 """Returns True if the apps.py file can access database records.
244 Arguments:
245 allow_test: If True, override checks and allow database access during testing mode
246 allow_plugins: If True, override checks and allow database access during plugin loading
247 allow_shell: If True, override checks and allow database access during shell sessions
249 There are some circumstances where we don't want the ready function in apps.py
250 to touch the database
251 """
252 # Prevent database access if we are running backups
253 if isRunningBackup(): 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true
254 return False
256 # Prevent database access if we are importing data
257 if not allow_plugins and isImportingData(): 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 return False
260 # Prevent database access if we are rebuilding data
261 if isRebuildingData(): 261 ↛ 262line 261 didn't jump to line 262 because the condition on line 261 was never true
262 return False
264 # Prevent database access if we are running migrations
265 if not allow_plugins and isRunningMigrations(): 265 ↛ 266line 265 didn't jump to line 266 because the condition on line 265 was never true
266 return False
268 # If any of the following management commands are being executed,
269 # prevent custom "on load" code from running!
270 excluded_commands = [
271 'compilemessages',
272 'createsuperuser',
273 'collectstatic',
274 'makemessages',
275 'spectactular',
276 'wait_for_db',
277 'check',
278 ]
280 if not allow_shell:
281 excluded_commands.append('shell')
283 if not allow_test:
284 # Override for testing mode?
285 excluded_commands.append('test')
287 if not allow_plugins:
288 excluded_commands.extend(['collectplugins'])
290 return all(cmd not in sys.argv for cmd in excluded_commands)
293def isPluginRegistryLoaded():
294 """Ensures that the plugin registry is already loaded.
296 The plugin registry reloads all apps onetime after starting if there are AppMixin plugins,
297 so that the discovered AppConfigs are added to Django. This triggers the ready function of
298 AppConfig to execute twice. Add this check to prevent from running two times.
300 Note: All apps using this check need to be registered after the plugins app in settings.py
302 Returns: 'False' if the registry has not fully loaded the plugins yet.
303 """
304 from plugin import registry
306 return registry.plugins_loaded
309def ignore_ready_warning(func):
310 """Decorator to ignore 'AppRegistryNotReady' warnings in functions called during app ready phase.
312 Ref: https://github.com/inventree/InvenTree/issues/10806
313 """
315 @functools.wraps(func)
316 def wrapper(*args, **kwargs):
317 with warnings.catch_warnings():
318 warnings.filterwarnings(
319 'ignore',
320 message='Accessing the database during app initialization is discouraged',
321 category=RuntimeWarning,
322 )
323 return func(*args, **kwargs)
325 return wrapper