inventree/InvenTree
2
0
Fork 0
mirror of https://github.com/inventree/InvenTree.git synced 2026-09-27 14:16:02 +00:00
Code Activity

evaluate enviroment once (#12890)

Browse Source
This commit is contained in:
Matthias Mair
2026-09-20 09:05:47 +10:00
committed by GitHub
parent ea553ea232
commit 63055865db
1 changed files with 140 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/InvenTree/InvenTree/ready.py
+152 -113
View File
@@ -35,65 +35,155 @@ def isAppLoaded(app_name: str) -> bool:
return app_name in _loaded_apps return app_name in _loaded_apps
def isInTestMode(): # Cached introspection of the command line arguments
"""Returns True if the database is in testing mode.""" ARGV_COMMANDS: dict[str, tuple[str, ...]] = {
return any(x in sys.argv for x in ['test', 'pytest']) or sys.argv[0].endswith( # Running in test mode
'pytest' 'test': ('test', 'pytest'),
) # The 'test' management command - not pytest
'test_command': ('test',),
# Collecting the available plugins
def isWaitingForDatabase(): 'collect_plugins': ('collectplugins',),
"""Return True if we are currently waiting for the database to be ready.""" # Listing the installed apps
return 'wait_for_db' in sys.argv 'list_apps': ('list_apps',),
# Running an interactive shell session
'shell': ('shell',),
def isImportingData(): # Running as a background worker
"""Returns True if the database is currently importing (or exporting) data, e.g. 'loaddata' command is performed.""" 'worker': ('qcluster',),
return any( # Running the django development server
x in sys.argv 'runserver': ('runserver',),
for x in ['flush', 'loaddata', 'bulkloaddata', 'dumpdata', 'bulkdumpdata'] # Waiting for the database to become available
) 'wait_for_db': ('wait_for_db',),
# #### #
# More complex groups
def isRunningMigrations(): # #### #
"""Return True if the database is currently running migrations.""" # Importing (or exporting) database records
return any( 'import_data': ('flush', 'loaddata', 'bulkloaddata', 'dumpdata', 'bulkdumpdata'),
x in sys.argv # Running database migrations
for x in ['migrate', 'makemigrations', 'showmigrations', 'runmigrations'] 'migrations': ('migrate', 'makemigrations', 'showmigrations', 'runmigrations'),
) # Rebuilding database records
'rebuild_data': (
def isRebuildingData():
"""Return true if any of the rebuilding commands are being executed."""
return any(
x in sys.argv
for x in [
'rebuild', 'rebuild',
'rebuild_models', 'rebuild_models',
'rebuild_thumbnails', 'rebuild_thumbnails',
'remove_stale_contenttypes', 'remove_stale_contenttypes',
] ),
) # Running a backup / restore operation
'backup': (
def isRunningBackup():
"""Return true if any of the backup commands are being executed."""
return any(
x in sys.argv
for x in [
'backup', 'backup',
'restore', 'restore',
'dbbackup', 'dbbackup',
'dbrestore', 'dbrestore',
'mediabackup', 'mediabackup',
'mediarestore', 'mediarestore',
] ),
) # Read-only commands, which should not trigger any database writes
'read_only': (
'help',
'check',
'shell',
'sqlflush',
'list_apps',
'wait_for_db',
'spectactular',
'makemessages',
'collectstatic',
'showmigrations',
'compilemessages',
),
# Commands which should *not* trigger schema generation
'schema_excluded': (
'compilemessages',
'createsuperuser',
'clean_settings',
'collectstatic',
'makemessages',
'wait_for_db',
'list_apps',
'gunicorn',
'sqlflush',
'qcluster',
'check',
'shell',
'help',
),
# Commands which *do* trigger schema generation
'schema_generation': (
'schema',
'spectactular',
# schema adjacent calls
'export_settings_definitions',
'export_tags',
'export_filters',
'export_report_context',
),
# Commands which must not touch the database during the app 'ready' phase
'database_excluded': (
'compilemessages',
'createsuperuser',
'collectstatic',
'makemessages',
'spectactular',
'wait_for_db',
'check',
),
}
def _introspectCommands(argv: list[str]):
"""Introspect the provided command line arguments."""
args = set(argv)
entrypoint = argv[0] if argv else ''
_context = {
key: not args.isdisjoint(commands) for key, commands in ARGV_COMMANDS.items()
}
# The entrypoint itself can indicate the context
_context['pytest_entrypoint'] = entrypoint.endswith('pytest')
_context['gunicorn_entrypoint'] = 'gunicorn' in entrypoint
# The development server is running without the auto-reloader
_context['noreload'] = '--noreload' in args
return _context
cmd_context = _introspectCommands(sys.argv)
def isInTestMode():
"""Returns True if the database is in testing mode."""
return cmd_context['test'] or cmd_context['pytest_entrypoint']
def isWaitingForDatabase():
"""Return True if we are currently waiting for the database to be ready."""
return cmd_context['wait_for_db']
def isImportingData():
"""Returns True if the database is currently importing (or exporting) data, e.g. 'loaddata' command is performed."""
return cmd_context['import_data']
def isRunningMigrations():
"""Return True if the database is currently running migrations."""
return cmd_context['migrations']
def isRebuildingData():
"""Return true if any of the rebuilding commands are being executed."""
return cmd_context['rebuild_data']
def isRunningBackup():
"""Return true if any of the backup commands are being executed."""
return cmd_context['backup']
def isCollectingPlugins(): def isCollectingPlugins():
"""Return True if the 'collectplugins' command is being executed.""" """Return True if the 'collectplugins' command is being executed."""
return 'collectplugins' in sys.argv return cmd_context['collect_plugins']
# This variable is used to cache the result of the isGeneratingSchema function, to prevent multiple executions of the same checks # This variable is used to cache the result of the isGeneratingSchema function, to prevent multiple executions of the same checks
@@ -133,36 +223,10 @@ def isGeneratingSchema():
return _setGeneratingSchema(False) return _setGeneratingSchema(False)
# Additional set of commands which should not trigger schema generation # Additional set of commands which should not trigger schema generation
excluded_commands = [ if cmd_context['schema_excluded']:
'compilemessages',
'createsuperuser',
'clean_settings',
'collectstatic',
'makemessages',
'wait_for_db',
'list_apps',
'gunicorn',
'sqlflush',
'qcluster',
'check',
'shell',
'help',
]
if any(cmd in sys.argv for cmd in excluded_commands):
return _setGeneratingSchema(False) return _setGeneratingSchema(False)
included_commands = [ if cmd_context['schema_generation']:
'schema',
'spectactular',
# schema adjacent calls
'export_settings_definitions',
'export_tags',
'export_filters',
'export_report_context',
]
if any(cmd in sys.argv for cmd in included_commands):
return _setGeneratingSchema(True) return _setGeneratingSchema(True)
# This is a very inefficient call - so we only use it as a last resort # This is a very inefficient call - so we only use it as a last resort
@@ -183,7 +247,7 @@ def isGeneratingSchema():
def isInWorkerThread(): def isInWorkerThread():
"""Returns True if the current thread is a background worker thread.""" """Returns True if the current thread is a background worker thread."""
return 'qcluster' in sys.argv return cmd_context['worker']
def isInServerThread(): def isInServerThread():
@@ -191,10 +255,10 @@ def isInServerThread():
if isInWorkerThread(): if isInWorkerThread():
return False return False
if 'runserver' in sys.argv: if cmd_context['runserver']:
return True return True
return 'gunicorn' in sys.argv[0] return cmd_context['gunicorn_entrypoint']
def isInMainThread(): def isInMainThread():
@@ -203,29 +267,12 @@ def isInMainThread():
- The RUN_MAIN env is set in that case. However if --noreload is applied, this variable - The RUN_MAIN env is set in that case. However if --noreload is applied, this variable
is not set because there are no different threads. is not set because there are no different threads.
""" """
if 'runserver' in sys.argv and '--noreload' not in sys.argv: if cmd_context['runserver'] and not cmd_context['noreload']:
return os.environ.get('RUN_MAIN', None) == 'true' return os.environ.get('RUN_MAIN', None) == 'true'
return not isInWorkerThread() return not isInWorkerThread()
def readOnlyCommands():
"""Return a list of read-only management commands which should not trigger database writes."""
return [
'help',
'check',
'shell',
'sqlflush',
'list_apps',
'wait_for_db',
'spectactular',
'makemessages',
'collectstatic',
'showmigrations',
'compilemessages',
]
def isReadOnlyCommand(): def isReadOnlyCommand():
"""Return True if the current command is a read-only command, which should not trigger any database writes.""" """Return True if the current command is a read-only command, which should not trigger any database writes."""
if ( if (
@@ -236,7 +283,7 @@ def isReadOnlyCommand():
): ):
return True return True
return any(cmd in sys.argv for cmd in readOnlyCommands()) return cmd_context['read_only']
def canAppAccessDatabase( def canAppAccessDatabase(
@@ -270,27 +317,19 @@ def canAppAccessDatabase(
# If any of the following management commands are being executed, # If any of the following management commands are being executed,
# prevent custom "on load" code from running! # prevent custom "on load" code from running!
excluded_commands = [ if cmd_context['database_excluded']:
'compilemessages', return False
'createsuperuser',
'collectstatic',
'makemessages',
'spectactular',
'wait_for_db',
'check',
]
if not allow_shell: if not allow_shell and cmd_context['shell']:
excluded_commands.append('shell') return False
if not allow_plugins and (
cmd_context['collect_plugins'] or cmd_context['list_apps']
):
return False
if not allow_test:
# Override for testing mode? # Override for testing mode?
excluded_commands.append('test') return allow_test or not cmd_context['test_command']
if not allow_plugins:
excluded_commands.extend(['collectplugins', 'list_apps'])
return all(cmd not in sys.argv for cmd in excluded_commands)
def isPluginRegistryLoaded(): def isPluginRegistryLoaded():