We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[FR] Support creating reading initial superuser account password from file
### Please verify that this feature request has NOT been suggested before.

- [X] I checked and didn't find a similar feature request

### Problem statement

I am running the docker-compose project on [NixOS](https://nixos.org/) through [Arion](https://docs.hercules-ci.com/arion/).

I am managing all of my secrets/passwords through [agenix](https://github.com/ryantm/agenix), which writes files containing the decrypted passwords to a specific path on the system (with appropriate permissions).
Due to the way Nix works, it is not safe to read the password file directly into the Arion config (Arion will generate a docker-compose.yaml file which will be readable to all users).

Instead, the recommended way for applications to ingest secrets is for them to be read at runtime.


### Suggested solution

I would like an option like `INVENTREE_ADMIN_PASSWORD_FILE` to be added to `add_user_on_startup()` which will read the contents of a file containing the password (mapped as a volume in the generated docker-compose.yaml)


### Describe alternatives you've considered

I could of course just forgo this feature and set up the superuser account manually, but part of the appeal of NixOS is the ability to bootstrap and configure the entire system with just a few commands

### Examples of other systems

From the [agenix readme](https://github.com/ryantm/agenix/blob/91220a701d0bf8055547620baa765b36d885db7a/README.md), shows the suggested way to access the secret (in this case defining linux user accounts)
```
{
  users.users.user1 = {
    isNormalUser = true;
    passwordFile = config.age.secrets.secret1.path;
  };
}
```

Instead of directly accessing a file, some NixOS modules instead take a generic shell command which prints the password to stdout, which gets run at the programs runtime:
```
  programs.msmtp = {
    enable = true;
    accounts.default = {
      auth = true;
      tls = true;
      tls_starttls = true;
      tls_trust_file = "/etc/ssl/certs/ca-bundle.crt";
      tls_certcheck = true;
      host = "smtp.office365.com";
      port = 587;
      from = "asdf@outlook.com";
      user = "asdf@outlook.com";
      passwordeval = "cat ${config.age.secrets.alert-outlook.path}";
    };
  };
```

### Do you want to develop this?

- [X] I want to develop this.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `InvenTree/InvenTree/apps.py`
Content:
```
1 """AppConfig for inventree app."""
2 
3 import logging
4 from importlib import import_module
5 from pathlib import Path
6 
7 from django.apps import AppConfig, apps
8 from django.conf import settings
9 from django.contrib.auth import get_user_model
10 from django.core.exceptions import AppRegistryNotReady
11 from django.db import transaction
12 from django.db.utils import IntegrityError, OperationalError
13 
14 import InvenTree.conversion
15 import InvenTree.tasks
16 from InvenTree.config import get_setting
17 from InvenTree.ready import (canAppAccessDatabase, isInMainThread,
18                              isInTestMode, isPluginRegistryLoaded)
19 
20 logger = logging.getLogger("inventree")
21 
22 
23 class InvenTreeConfig(AppConfig):
24     """AppConfig for inventree app."""
25     name = 'InvenTree'
26 
27     def ready(self):
28         """Run system wide setup init steps.
29 
30         Like:
31         - Checking if migrations should be run
32         - Cleaning up tasks
33         - Starting regular tasks
34         - Updating exchange rates
35         - Collecting notification methods
36         - Collecting state transition methods
37         - Adding users set in the current environment
38         """
39         # skip loading if plugin registry is not loaded or we run in a background thread
40         if not isPluginRegistryLoaded() or not isInMainThread():
41             return
42 
43         if canAppAccessDatabase() or settings.TESTING_ENV:
44 
45             self.remove_obsolete_tasks()
46 
47             self.collect_tasks()
48             self.start_background_tasks()
49 
50             if not isInTestMode():  # pragma: no cover
51                 self.update_exchange_rates()
52                 # Let the background worker check for migrations
53                 InvenTree.tasks.offload_task(InvenTree.tasks.check_for_migrations)
54 
55         self.collect_notification_methods()
56         self.collect_state_transition_methods()
57 
58         # Ensure the unit registry is loaded
59         InvenTree.conversion.get_unit_registry()
60 
61         if canAppAccessDatabase() or settings.TESTING_ENV:
62             self.add_user_on_startup()
63 
64     def remove_obsolete_tasks(self):
65         """Delete any obsolete scheduled tasks in the database."""
66         obsolete = [
67             'InvenTree.tasks.delete_expired_sessions',
68             'stock.tasks.delete_old_stock_items',
69         ]
70 
71         try:
72             from django_q.models import Schedule
73         except AppRegistryNotReady:  # pragma: no cover
74             return
75 
76         # Remove any existing obsolete tasks
77         try:
78             Schedule.objects.filter(func__in=obsolete).delete()
79         except Exception:
80             logger.exception("Failed to remove obsolete tasks - database not ready")
81 
82     def start_background_tasks(self):
83         """Start all background tests for InvenTree."""
84         logger.info("Starting background tasks...")
85 
86         from django_q.models import Schedule
87 
88         # List of existing scheduled tasks (in the database)
89         existing_tasks = {}
90 
91         for existing_task in Schedule.objects.all():
92             existing_tasks[existing_task.func] = existing_task
93 
94         tasks_to_create = []
95         tasks_to_update = []
96 
97         # List of collected tasks found with the @scheduled_task decorator
98         tasks = InvenTree.tasks.tasks.task_list
99 
100         for task in tasks:
101 
102             ref_name = f'{task.func.__module__}.{task.func.__name__}'
103 
104             if ref_name in existing_tasks.keys():
105                 # This task already exists - update the details if required
106                 existing_task = existing_tasks[ref_name]
107 
108                 if existing_task.schedule_type != task.interval or existing_task.minutes != task.minutes:
109 
110                     existing_task.schedule_type = task.interval
111                     existing_task.minutes = task.minutes
112                     tasks_to_update.append(existing_task)
113 
114             else:
115                 # This task does *not* already exist - create it
116                 tasks_to_create.append(
117                     Schedule(
118                         name=ref_name,
119                         func=ref_name,
120                         schedule_type=task.interval,
121                         minutes=task.minutes,
122                     )
123                 )
124 
125         if len(tasks_to_create) > 0:
126             Schedule.objects.bulk_create(tasks_to_create)
127             logger.info("Created %s new scheduled tasks", len(tasks_to_create))
128 
129         if len(tasks_to_update) > 0:
130             Schedule.objects.bulk_update(tasks_to_update, ['schedule_type', 'minutes'])
131             logger.info("Updated %s existing scheduled tasks", len(tasks_to_update))
132 
133         # Put at least one task onto the background worker stack,
134         # which will be processed as soon as the worker comes online
135         InvenTree.tasks.offload_task(
136             InvenTree.tasks.heartbeat,
137             force_async=True,
138         )
139 
140         logger.info("Started %s scheduled background tasks...", len(tasks))
141 
142     def collect_tasks(self):
143         """Collect all background tasks."""
144         for app_name, app in apps.app_configs.items():
145             if app_name == 'InvenTree':
146                 continue
147 
148             if Path(app.path).joinpath('tasks.py').exists():
149                 try:
150                     import_module(f'{app.module.__package__}.tasks')
151                 except Exception as e:  # pragma: no cover
152                     logger.exception("Error loading tasks for %s: %s", app_name, e)
153 
154     def update_exchange_rates(self):  # pragma: no cover
155         """Update exchange rates each time the server is started.
156 
157         Only runs *if*:
158         a) Have not been updated recently (one day or less)
159         b) The base exchange rate has been altered
160         """
161         try:
162             from djmoney.contrib.exchange.models import ExchangeBackend
163 
164             from common.settings import currency_code_default
165             from InvenTree.tasks import update_exchange_rates
166         except AppRegistryNotReady:  # pragma: no cover
167             pass
168 
169         base_currency = currency_code_default()
170 
171         update = False
172 
173         try:
174             backend = ExchangeBackend.objects.filter(name='InvenTreeExchange')
175 
176             if backend.exists():
177                 backend = backend.first()
178 
179                 last_update = backend.last_update
180 
181                 if last_update is None:
182                     # Never been updated
183                     logger.info("Exchange backend has never been updated")
184                     update = True
185 
186                 # Backend currency has changed?
187                 if base_currency != backend.base_currency:
188                     logger.info("Base currency changed from %s to %s", backend.base_currency, base_currency)
189                     update = True
190 
191         except (ExchangeBackend.DoesNotExist):
192             logger.info("Exchange backend not found - updating")
193             update = True
194 
195         except Exception:
196             # Some other error - potentially the tables are not ready yet
197             return
198 
199         if update:
200             try:
201                 update_exchange_rates()
202             except OperationalError:
203                 logger.warning("Could not update exchange rates - database not ready")
204             except Exception as e:
205                 logger.exception("Error updating exchange rates: %s (%s)", e, type(e))
206 
207     def add_user_on_startup(self):
208         """Add a user on startup."""
209         # stop if checks were already created
210         if hasattr(settings, 'USER_ADDED') and settings.USER_ADDED:
211             return
212 
213         # get values
214         add_user = get_setting('INVENTREE_ADMIN_USER', 'admin_user')
215         add_email = get_setting('INVENTREE_ADMIN_EMAIL', 'admin_email')
216         add_password = get_setting('INVENTREE_ADMIN_PASSWORD', 'admin_password')
217 
218         # check if all values are present
219         set_variables = 0
220 
221         for tested_var in [add_user, add_email, add_password]:
222             if tested_var:
223                 set_variables += 1
224 
225         # no variable set -> do not try anything
226         if set_variables == 0:
227             settings.USER_ADDED = True
228             return
229 
230         # not all needed variables set
231         if set_variables < 3:
232             logger.warning('Not all required settings for adding a user on startup are present:\nINVENTREE_ADMIN_USER, INVENTREE_ADMIN_EMAIL, INVENTREE_ADMIN_PASSWORD')
233             settings.USER_ADDED = True
234             return
235 
236         # good to go -> create user
237         user = get_user_model()
238         try:
239             with transaction.atomic():
240                 if user.objects.filter(username=add_user).exists():
241                     logger.info("User %s already exists - skipping creation", add_user)
242                 else:
243                     new_user = user.objects.create_superuser(add_user, add_email, add_password)
244                     logger.info('User %s was created!', str(new_user))
245         except IntegrityError:
246             logger.warning('The user "%s" could not be created', add_user)
247 
248         # do not try again
249         settings.USER_ADDED = True
250 
251     def collect_notification_methods(self):
252         """Collect all notification methods."""
253         from common.notifications import storage
254 
255         storage.collect()
256 
257     def collect_state_transition_methods(self):
258         """Collect all state transition methods."""
259         from generic.states import storage
260 
261         storage.collect()
262
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/InvenTree/InvenTree/apps.py b/InvenTree/InvenTree/apps.py
--- a/InvenTree/InvenTree/apps.py
+++ b/InvenTree/InvenTree/apps.py
@@ -60,6 +60,7 @@
 
         if canAppAccessDatabase() or settings.TESTING_ENV:
             self.add_user_on_startup()
+            self.add_user_from_file()
 
     def remove_obsolete_tasks(self):
         """Delete any obsolete scheduled tasks in the database."""
@@ -214,6 +215,7 @@
         add_user = get_setting('INVENTREE_ADMIN_USER', 'admin_user')
         add_email = get_setting('INVENTREE_ADMIN_EMAIL', 'admin_email')
         add_password = get_setting('INVENTREE_ADMIN_PASSWORD', 'admin_password')
+        add_password_file = get_setting("INVENTREE_ADMIN_PASSWORD_FILE", "admin_password_file", None)
 
         # check if all values are present
         set_variables = 0
@@ -229,11 +231,21 @@
 
         # not all needed variables set
         if set_variables < 3:
-            logger.warning('Not all required settings for adding a user on startup are present:\nINVENTREE_ADMIN_USER, INVENTREE_ADMIN_EMAIL, INVENTREE_ADMIN_PASSWORD')
             settings.USER_ADDED = True
+
+            # if a password file is present, do not warn - will be handled later
+            if add_password_file:
+                return
+            logger.warning('Not all required settings for adding a user on startup are present:\nINVENTREE_ADMIN_USER, INVENTREE_ADMIN_EMAIL, INVENTREE_ADMIN_PASSWORD')
             return
 
         # good to go -> create user
+        self._create_admin_user(add_user, add_email, add_password)
+
+        # do not try again
+        settings.USER_ADDED = True
+
+    def _create_admin_user(self, add_user, add_email, add_password):
         user = get_user_model()
         try:
             with transaction.atomic():
@@ -245,8 +257,34 @@
         except IntegrityError:
             logger.warning('The user "%s" could not be created', add_user)
 
+    def add_user_from_file(self):
+        """Add the superuser from a file."""
+        # stop if checks were already created
+        if hasattr(settings, "USER_ADDED_FILE") and settings.USER_ADDED_FILE:
+            return
+
+        # get values
+        add_password_file = get_setting(
+            "INVENTREE_ADMIN_PASSWORD_FILE", "admin_password_file", None
+        )
+
+        # no variable set -> do not try anything
+        if not add_password_file:
+            settings.USER_ADDED_FILE = True
+            return
+
+        # check if file exists
+        add_password_file = Path(str(add_password_file))
+        if not add_password_file.exists():
+            logger.warning('The file "%s" does not exist', add_password_file)
+            settings.USER_ADDED_FILE = True
+            return
+
+        # good to go -> create user
+        self._create_admin_user(get_setting('INVENTREE_ADMIN_USER', 'admin_user', 'admin'), get_setting('INVENTREE_ADMIN_EMAIL', 'admin_email', ''), add_password_file.read_text(encoding="utf-8"))
+
         # do not try again
-        settings.USER_ADDED = True
+        settings.USER_ADDED_FILE = True
 
     def collect_notification_methods(self):
         """Collect all notification methods."""

{"golden_diff": "diff --git a/InvenTree/InvenTree/apps.py b/InvenTree/InvenTree/apps.py\n--- a/InvenTree/InvenTree/apps.py\n+++ b/InvenTree/InvenTree/apps.py\n@@ -60,6 +60,7 @@\n \n         if canAppAccessDatabase() or settings.TESTING_ENV:\n             self.add_user_on_startup()\n+            self.add_user_from_file()\n \n     def remove_obsolete_tasks(self):\n         \"\"\"Delete any obsolete scheduled tasks in the database.\"\"\"\n@@ -214,6 +215,7 @@\n         add_user = get_setting('INVENTREE_ADMIN_USER', 'admin_user')\n         add_email = get_setting('INVENTREE_ADMIN_EMAIL', 'admin_email')\n         add_password = get_setting('INVENTREE_ADMIN_PASSWORD', 'admin_password')\n+        add_password_file = get_setting(\"INVENTREE_ADMIN_PASSWORD_FILE\", \"admin_password_file\", None)\n \n         # check if all values are present\n         set_variables = 0\n@@ -229,11 +231,21 @@\n \n         # not all needed variables set\n         if set_variables < 3:\n-            logger.warning('Not all required settings for adding a user on startup are present:\\nINVENTREE_ADMIN_USER, INVENTREE_ADMIN_EMAIL, INVENTREE_ADMIN_PASSWORD')\n             settings.USER_ADDED = True\n+\n+            # if a password file is present, do not warn - will be handled later\n+            if add_password_file:\n+                return\n+            logger.warning('Not all required settings for adding a user on startup are present:\\nINVENTREE_ADMIN_USER, INVENTREE_ADMIN_EMAIL, INVENTREE_ADMIN_PASSWORD')\n             return\n \n         # good to go -> create user\n+        self._create_admin_user(add_user, add_email, add_password)\n+\n+        # do not try again\n+        settings.USER_ADDED = True\n+\n+    def _create_admin_user(self, add_user, add_email, add_password):\n         user = get_user_model()\n         try:\n             with transaction.atomic():\n@@ -245,8 +257,34 @@\n         except IntegrityError:\n             logger.warning('The user \"%s\" could not be created', add_user)\n \n+    def add_user_from_file(self):\n+        \"\"\"Add the superuser from a file.\"\"\"\n+        # stop if checks were already created\n+        if hasattr(settings, \"USER_ADDED_FILE\") and settings.USER_ADDED_FILE:\n+            return\n+\n+        # get values\n+        add_password_file = get_setting(\n+            \"INVENTREE_ADMIN_PASSWORD_FILE\", \"admin_password_file\", None\n+        )\n+\n+        # no variable set -> do not try anything\n+        if not add_password_file:\n+            settings.USER_ADDED_FILE = True\n+            return\n+\n+        # check if file exists\n+        add_password_file = Path(str(add_password_file))\n+        if not add_password_file.exists():\n+            logger.warning('The file \"%s\" does not exist', add_password_file)\n+            settings.USER_ADDED_FILE = True\n+            return\n+\n+        # good to go -> create user\n+        self._create_admin_user(get_setting('INVENTREE_ADMIN_USER', 'admin_user', 'admin'), get_setting('INVENTREE_ADMIN_EMAIL', 'admin_email', ''), add_password_file.read_text(encoding=\"utf-8\"))\n+\n         # do not try again\n-        settings.USER_ADDED = True\n+        settings.USER_ADDED_FILE = True\n \n     def collect_notification_methods(self):\n         \"\"\"Collect all notification methods.\"\"\"\n", "issue": "[FR] Support creating reading initial superuser account password from file\n### Please verify that this feature request has NOT been suggested before.\n\n- [X] I checked and didn't find a similar feature request\n\n### Problem statement\n\nI am running the docker-compose project on [NixOS](https://nixos.org/) through [Arion](https://docs.hercules-ci.com/arion/).\r\n\r\nI am managing all of my secrets/passwords through [agenix](https://github.com/ryantm/agenix), which writes files containing the decrypted passwords to a specific path on the system (with appropriate permissions).\r\nDue to the way Nix works, it is not safe to read the password file directly into the Arion config (Arion will generate a docker-compose.yaml file which will be readable to all users).\r\n\r\nInstead, the recommended way for applications to ingest secrets is for them to be read at runtime.\r\n\n\n### Suggested solution\n\nI would like an option like `INVENTREE_ADMIN_PASSWORD_FILE` to be added to `add_user_on_startup()` which will read the contents of a file containing the password (mapped as a volume in the generated docker-compose.yaml)\r\n\n\n### Describe alternatives you've considered\n\nI could of course just forgo this feature and set up the superuser account manually, but part of the appeal of NixOS is the ability to bootstrap and configure the entire system with just a few commands\n\n### Examples of other systems\n\nFrom the [agenix readme](https://github.com/ryantm/agenix/blob/91220a701d0bf8055547620baa765b36d885db7a/README.md), shows the suggested way to access the secret (in this case defining linux user accounts)\r\n```\r\n{\r\n  users.users.user1 = {\r\n    isNormalUser = true;\r\n    passwordFile = config.age.secrets.secret1.path;\r\n  };\r\n}\r\n```\r\n\r\nInstead of directly accessing a file, some NixOS modules instead take a generic shell command which prints the password to stdout, which gets run at the programs runtime:\r\n```\r\n  programs.msmtp = {\r\n    enable = true;\r\n    accounts.default = {\r\n      auth = true;\r\n      tls = true;\r\n      tls_starttls = true;\r\n      tls_trust_file = \"/etc/ssl/certs/ca-bundle.crt\";\r\n      tls_certcheck = true;\r\n      host = \"smtp.office365.com\";\r\n      port = 587;\r\n      from = \"asdf@outlook.com\";\r\n      user = \"asdf@outlook.com\";\r\n      passwordeval = \"cat ${config.age.secrets.alert-outlook.path}\";\r\n    };\r\n  };\r\n```\n\n### Do you want to develop this?\n\n- [X] I want to develop this.\n", "before_files": [{"content": "\"\"\"AppConfig for inventree app.\"\"\"\n\nimport logging\nfrom importlib import import_module\nfrom pathlib import Path\n\nfrom django.apps import AppConfig, apps\nfrom django.conf import settings\nfrom django.contrib.auth import get_user_model\nfrom django.core.exceptions import AppRegistryNotReady\nfrom django.db import transaction\nfrom django.db.utils import IntegrityError, OperationalError\n\nimport InvenTree.conversion\nimport InvenTree.tasks\nfrom InvenTree.config import get_setting\nfrom InvenTree.ready import (canAppAccessDatabase, isInMainThread,\n                             isInTestMode, isPluginRegistryLoaded)\n\nlogger = logging.getLogger(\"inventree\")\n\n\nclass InvenTreeConfig(AppConfig):\n    \"\"\"AppConfig for inventree app.\"\"\"\n    name = 'InvenTree'\n\n    def ready(self):\n        \"\"\"Run system wide setup init steps.\n\n        Like:\n        - Checking if migrations should be run\n        - Cleaning up tasks\n        - Starting regular tasks\n        - Updating exchange rates\n        - Collecting notification methods\n        - Collecting state transition methods\n        - Adding users set in the current environment\n        \"\"\"\n        # skip loading if plugin registry is not loaded or we run in a background thread\n        if not isPluginRegistryLoaded() or not isInMainThread():\n            return\n\n        if canAppAccessDatabase() or settings.TESTING_ENV:\n\n            self.remove_obsolete_tasks()\n\n            self.collect_tasks()\n            self.start_background_tasks()\n\n            if not isInTestMode():  # pragma: no cover\n                self.update_exchange_rates()\n                # Let the background worker check for migrations\n                InvenTree.tasks.offload_task(InvenTree.tasks.check_for_migrations)\n\n        self.collect_notification_methods()\n        self.collect_state_transition_methods()\n\n        # Ensure the unit registry is loaded\n        InvenTree.conversion.get_unit_registry()\n\n        if canAppAccessDatabase() or settings.TESTING_ENV:\n            self.add_user_on_startup()\n\n    def remove_obsolete_tasks(self):\n        \"\"\"Delete any obsolete scheduled tasks in the database.\"\"\"\n        obsolete = [\n            'InvenTree.tasks.delete_expired_sessions',\n            'stock.tasks.delete_old_stock_items',\n        ]\n\n        try:\n            from django_q.models import Schedule\n        except AppRegistryNotReady:  # pragma: no cover\n            return\n\n        # Remove any existing obsolete tasks\n        try:\n            Schedule.objects.filter(func__in=obsolete).delete()\n        except Exception:\n            logger.exception(\"Failed to remove obsolete tasks - database not ready\")\n\n    def start_background_tasks(self):\n        \"\"\"Start all background tests for InvenTree.\"\"\"\n        logger.info(\"Starting background tasks...\")\n\n        from django_q.models import Schedule\n\n        # List of existing scheduled tasks (in the database)\n        existing_tasks = {}\n\n        for existing_task in Schedule.objects.all():\n            existing_tasks[existing_task.func] = existing_task\n\n        tasks_to_create = []\n        tasks_to_update = []\n\n        # List of collected tasks found with the @scheduled_task decorator\n        tasks = InvenTree.tasks.tasks.task_list\n\n        for task in tasks:\n\n            ref_name = f'{task.func.__module__}.{task.func.__name__}'\n\n            if ref_name in existing_tasks.keys():\n                # This task already exists - update the details if required\n                existing_task = existing_tasks[ref_name]\n\n                if existing_task.schedule_type != task.interval or existing_task.minutes != task.minutes:\n\n                    existing_task.schedule_type = task.interval\n                    existing_task.minutes = task.minutes\n                    tasks_to_update.append(existing_task)\n\n            else:\n                # This task does *not* already exist - create it\n                tasks_to_create.append(\n                    Schedule(\n                        name=ref_name,\n                        func=ref_name,\n                        schedule_type=task.interval,\n                        minutes=task.minutes,\n                    )\n                )\n\n        if len(tasks_to_create) > 0:\n            Schedule.objects.bulk_create(tasks_to_create)\n            logger.info(\"Created %s new scheduled tasks\", len(tasks_to_create))\n\n        if len(tasks_to_update) > 0:\n            Schedule.objects.bulk_update(tasks_to_update, ['schedule_type', 'minutes'])\n            logger.info(\"Updated %s existing scheduled tasks\", len(tasks_to_update))\n\n        # Put at least one task onto the background worker stack,\n        # which will be processed as soon as the worker comes online\n        InvenTree.tasks.offload_task(\n            InvenTree.tasks.heartbeat,\n            force_async=True,\n        )\n\n        logger.info(\"Started %s scheduled background tasks...\", len(tasks))\n\n    def collect_tasks(self):\n        \"\"\"Collect all background tasks.\"\"\"\n        for app_name, app in apps.app_configs.items():\n            if app_name == 'InvenTree':\n                continue\n\n            if Path(app.path).joinpath('tasks.py').exists():\n                try:\n                    import_module(f'{app.module.__package__}.tasks')\n                except Exception as e:  # pragma: no cover\n                    logger.exception(\"Error loading tasks for %s: %s\", app_name, e)\n\n    def update_exchange_rates(self):  # pragma: no cover\n        \"\"\"Update exchange rates each time the server is started.\n\n        Only runs *if*:\n        a) Have not been updated recently (one day or less)\n        b) The base exchange rate has been altered\n        \"\"\"\n        try:\n            from djmoney.contrib.exchange.models import ExchangeBackend\n\n            from common.settings import currency_code_default\n            from InvenTree.tasks import update_exchange_rates\n        except AppRegistryNotReady:  # pragma: no cover\n            pass\n\n        base_currency = currency_code_default()\n\n        update = False\n\n        try:\n            backend = ExchangeBackend.objects.filter(name='InvenTreeExchange')\n\n            if backend.exists():\n                backend = backend.first()\n\n                last_update = backend.last_update\n\n                if last_update is None:\n                    # Never been updated\n                    logger.info(\"Exchange backend has never been updated\")\n                    update = True\n\n                # Backend currency has changed?\n                if base_currency != backend.base_currency:\n                    logger.info(\"Base currency changed from %s to %s\", backend.base_currency, base_currency)\n                    update = True\n\n        except (ExchangeBackend.DoesNotExist):\n            logger.info(\"Exchange backend not found - updating\")\n            update = True\n\n        except Exception:\n            # Some other error - potentially the tables are not ready yet\n            return\n\n        if update:\n            try:\n                update_exchange_rates()\n            except OperationalError:\n                logger.warning(\"Could not update exchange rates - database not ready\")\n            except Exception as e:\n                logger.exception(\"Error updating exchange rates: %s (%s)\", e, type(e))\n\n    def add_user_on_startup(self):\n        \"\"\"Add a user on startup.\"\"\"\n        # stop if checks were already created\n        if hasattr(settings, 'USER_ADDED') and settings.USER_ADDED:\n            return\n\n        # get values\n        add_user = get_setting('INVENTREE_ADMIN_USER', 'admin_user')\n        add_email = get_setting('INVENTREE_ADMIN_EMAIL', 'admin_email')\n        add_password = get_setting('INVENTREE_ADMIN_PASSWORD', 'admin_password')\n\n        # check if all values are present\n        set_variables = 0\n\n        for tested_var in [add_user, add_email, add_password]:\n            if tested_var:\n                set_variables += 1\n\n        # no variable set -> do not try anything\n        if set_variables == 0:\n            settings.USER_ADDED = True\n            return\n\n        # not all needed variables set\n        if set_variables < 3:\n            logger.warning('Not all required settings for adding a user on startup are present:\\nINVENTREE_ADMIN_USER, INVENTREE_ADMIN_EMAIL, INVENTREE_ADMIN_PASSWORD')\n            settings.USER_ADDED = True\n            return\n\n        # good to go -> create user\n        user = get_user_model()\n        try:\n            with transaction.atomic():\n                if user.objects.filter(username=add_user).exists():\n                    logger.info(\"User %s already exists - skipping creation\", add_user)\n                else:\n                    new_user = user.objects.create_superuser(add_user, add_email, add_password)\n                    logger.info('User %s was created!', str(new_user))\n        except IntegrityError:\n            logger.warning('The user \"%s\" could not be created', add_user)\n\n        # do not try again\n        settings.USER_ADDED = True\n\n    def collect_notification_methods(self):\n        \"\"\"Collect all notification methods.\"\"\"\n        from common.notifications import storage\n\n        storage.collect()\n\n    def collect_state_transition_methods(self):\n        \"\"\"Collect all state transition methods.\"\"\"\n        from generic.states import storage\n\n        storage.collect()\n", "path": "InvenTree/InvenTree/apps.py"}], "after_files": [{"content": "\"\"\"AppConfig for inventree app.\"\"\"\n\nimport logging\nfrom importlib import import_module\nfrom pathlib import Path\n\nfrom django.apps import AppConfig, apps\nfrom django.conf import settings\nfrom django.contrib.auth import get_user_model\nfrom django.core.exceptions import AppRegistryNotReady\nfrom django.db import transaction\nfrom django.db.utils import IntegrityError, OperationalError\n\nimport InvenTree.conversion\nimport InvenTree.tasks\nfrom InvenTree.config import get_setting\nfrom InvenTree.ready import (canAppAccessDatabase, isInMainThread,\n                             isInTestMode, isPluginRegistryLoaded)\n\nlogger = logging.getLogger(\"inventree\")\n\n\nclass InvenTreeConfig(AppConfig):\n    \"\"\"AppConfig for inventree app.\"\"\"\n    name = 'InvenTree'\n\n    def ready(self):\n        \"\"\"Run system wide setup init steps.\n\n        Like:\n        - Checking if migrations should be run\n        - Cleaning up tasks\n        - Starting regular tasks\n        - Updating exchange rates\n        - Collecting notification methods\n        - Collecting state transition methods\n        - Adding users set in the current environment\n        \"\"\"\n        # skip loading if plugin registry is not loaded or we run in a background thread\n        if not isPluginRegistryLoaded() or not isInMainThread():\n            return\n\n        if canAppAccessDatabase() or settings.TESTING_ENV:\n\n            self.remove_obsolete_tasks()\n\n            self.collect_tasks()\n            self.start_background_tasks()\n\n            if not isInTestMode():  # pragma: no cover\n                self.update_exchange_rates()\n                # Let the background worker check for migrations\n                InvenTree.tasks.offload_task(InvenTree.tasks.check_for_migrations)\n\n        self.collect_notification_methods()\n        self.collect_state_transition_methods()\n\n        # Ensure the unit registry is loaded\n        InvenTree.conversion.get_unit_registry()\n\n        if canAppAccessDatabase() or settings.TESTING_ENV:\n            self.add_user_on_startup()\n            self.add_user_from_file()\n\n    def remove_obsolete_tasks(self):\n        \"\"\"Delete any obsolete scheduled tasks in the database.\"\"\"\n        obsolete = [\n            'InvenTree.tasks.delete_expired_sessions',\n            'stock.tasks.delete_old_stock_items',\n        ]\n\n        try:\n            from django_q.models import Schedule\n        except AppRegistryNotReady:  # pragma: no cover\n            return\n\n        # Remove any existing obsolete tasks\n        try:\n            Schedule.objects.filter(func__in=obsolete).delete()\n        except Exception:\n            logger.exception(\"Failed to remove obsolete tasks - database not ready\")\n\n    def start_background_tasks(self):\n        \"\"\"Start all background tests for InvenTree.\"\"\"\n        logger.info(\"Starting background tasks...\")\n\n        from django_q.models import Schedule\n\n        # List of existing scheduled tasks (in the database)\n        existing_tasks = {}\n\n        for existing_task in Schedule.objects.all():\n            existing_tasks[existing_task.func] = existing_task\n\n        tasks_to_create = []\n        tasks_to_update = []\n\n        # List of collected tasks found with the @scheduled_task decorator\n        tasks = InvenTree.tasks.tasks.task_list\n\n        for task in tasks:\n\n            ref_name = f'{task.func.__module__}.{task.func.__name__}'\n\n            if ref_name in existing_tasks.keys():\n                # This task already exists - update the details if required\n                existing_task = existing_tasks[ref_name]\n\n                if existing_task.schedule_type != task.interval or existing_task.minutes != task.minutes:\n\n                    existing_task.schedule_type = task.interval\n                    existing_task.minutes = task.minutes\n                    tasks_to_update.append(existing_task)\n\n            else:\n                # This task does *not* already exist - create it\n                tasks_to_create.append(\n                    Schedule(\n                        name=ref_name,\n                        func=ref_name,\n                        schedule_type=task.interval,\n                        minutes=task.minutes,\n                    )\n                )\n\n        if len(tasks_to_create) > 0:\n            Schedule.objects.bulk_create(tasks_to_create)\n            logger.info(\"Created %s new scheduled tasks\", len(tasks_to_create))\n\n        if len(tasks_to_update) > 0:\n            Schedule.objects.bulk_update(tasks_to_update, ['schedule_type', 'minutes'])\n            logger.info(\"Updated %s existing scheduled tasks\", len(tasks_to_update))\n\n        # Put at least one task onto the background worker stack,\n        # which will be processed as soon as the worker comes online\n        InvenTree.tasks.offload_task(\n            InvenTree.tasks.heartbeat,\n            force_async=True,\n        )\n\n        logger.info(\"Started %s scheduled background tasks...\", len(tasks))\n\n    def collect_tasks(self):\n        \"\"\"Collect all background tasks.\"\"\"\n        for app_name, app in apps.app_configs.items():\n            if app_name == 'InvenTree':\n                continue\n\n            if Path(app.path).joinpath('tasks.py').exists():\n                try:\n                    import_module(f'{app.module.__package__}.tasks')\n                except Exception as e:  # pragma: no cover\n                    logger.exception(\"Error loading tasks for %s: %s\", app_name, e)\n\n    def update_exchange_rates(self):  # pragma: no cover\n        \"\"\"Update exchange rates each time the server is started.\n\n        Only runs *if*:\n        a) Have not been updated recently (one day or less)\n        b) The base exchange rate has been altered\n        \"\"\"\n        try:\n            from djmoney.contrib.exchange.models import ExchangeBackend\n\n            from common.settings import currency_code_default\n            from InvenTree.tasks import update_exchange_rates\n        except AppRegistryNotReady:  # pragma: no cover\n            pass\n\n        base_currency = currency_code_default()\n\n        update = False\n\n        try:\n            backend = ExchangeBackend.objects.filter(name='InvenTreeExchange')\n\n            if backend.exists():\n                backend = backend.first()\n\n                last_update = backend.last_update\n\n                if last_update is None:\n                    # Never been updated\n                    logger.info(\"Exchange backend has never been updated\")\n                    update = True\n\n                # Backend currency has changed?\n                if base_currency != backend.base_currency:\n                    logger.info(\"Base currency changed from %s to %s\", backend.base_currency, base_currency)\n                    update = True\n\n        except (ExchangeBackend.DoesNotExist):\n            logger.info(\"Exchange backend not found - updating\")\n            update = True\n\n        except Exception:\n            # Some other error - potentially the tables are not ready yet\n            return\n\n        if update:\n            try:\n                update_exchange_rates()\n            except OperationalError:\n                logger.warning(\"Could not update exchange rates - database not ready\")\n            except Exception as e:\n                logger.exception(\"Error updating exchange rates: %s (%s)\", e, type(e))\n\n    def add_user_on_startup(self):\n        \"\"\"Add a user on startup.\"\"\"\n        # stop if checks were already created\n        if hasattr(settings, 'USER_ADDED') and settings.USER_ADDED:\n            return\n\n        # get values\n        add_user = get_setting('INVENTREE_ADMIN_USER', 'admin_user')\n        add_email = get_setting('INVENTREE_ADMIN_EMAIL', 'admin_email')\n        add_password = get_setting('INVENTREE_ADMIN_PASSWORD', 'admin_password')\n        add_password_file = get_setting(\"INVENTREE_ADMIN_PASSWORD_FILE\", \"admin_password_file\", None)\n\n        # check if all values are present\n        set_variables = 0\n\n        for tested_var in [add_user, add_email, add_password]:\n            if tested_var:\n                set_variables += 1\n\n        # no variable set -> do not try anything\n        if set_variables == 0:\n            settings.USER_ADDED = True\n            return\n\n        # not all needed variables set\n        if set_variables < 3:\n            settings.USER_ADDED = True\n\n            # if a password file is present, do not warn - will be handled later\n            if add_password_file:\n                return\n            logger.warning('Not all required settings for adding a user on startup are present:\\nINVENTREE_ADMIN_USER, INVENTREE_ADMIN_EMAIL, INVENTREE_ADMIN_PASSWORD')\n            return\n\n        # good to go -> create user\n        self._create_admin_user(add_user, add_email, add_password)\n\n        # do not try again\n        settings.USER_ADDED = True\n\n    def _create_admin_user(self, add_user, add_email, add_password):\n        user = get_user_model()\n        try:\n            with transaction.atomic():\n                if user.objects.filter(username=add_user).exists():\n                    logger.info(\"User %s already exists - skipping creation\", add_user)\n                else:\n                    new_user = user.objects.create_superuser(add_user, add_email, add_password)\n                    logger.info('User %s was created!', str(new_user))\n        except IntegrityError:\n            logger.warning('The user \"%s\" could not be created', add_user)\n\n    def add_user_from_file(self):\n        \"\"\"Add the superuser from a file.\"\"\"\n        # stop if checks were already created\n        if hasattr(settings, \"USER_ADDED_FILE\") and settings.USER_ADDED_FILE:\n            return\n\n        # get values\n        add_password_file = get_setting(\n            \"INVENTREE_ADMIN_PASSWORD_FILE\", \"admin_password_file\", None\n        )\n\n        # no variable set -> do not try anything\n        if not add_password_file:\n            settings.USER_ADDED_FILE = True\n            return\n\n        # check if file exists\n        add_password_file = Path(str(add_password_file))\n        if not add_password_file.exists():\n            logger.warning('The file \"%s\" does not exist', add_password_file)\n            settings.USER_ADDED_FILE = True\n            return\n\n        # good to go -> create user\n        self._create_admin_user(get_setting('INVENTREE_ADMIN_USER', 'admin_user', 'admin'), get_setting('INVENTREE_ADMIN_EMAIL', 'admin_email', ''), add_password_file.read_text(encoding=\"utf-8\"))\n\n        # do not try again\n        settings.USER_ADDED_FILE = True\n\n    def collect_notification_methods(self):\n        \"\"\"Collect all notification methods.\"\"\"\n        from common.notifications import storage\n\n        storage.collect()\n\n    def collect_state_transition_methods(self):\n        \"\"\"Collect all state transition methods.\"\"\"\n        from generic.states import storage\n\n        storage.collect()\n", "path": "InvenTree/InvenTree/apps.py"}]}