Reorganize management commands

netbox/core/management/commands/clearcache.py

@@ -0,0 +1,11 @@
+from django.core.cache import cache
+from django.core.management.base import BaseCommand
+
+
+class Command(BaseCommand):
+    """Command to clear the entire cache."""
+    help = 'Clears the cache.'
+
+    def handle(self, *args, **kwargs):
+        cache.clear()
+        self.stdout.write('Cache has been cleared.', ending="\n")

netbox/core/management/commands/makemigrations.py

@@ -0,0 +1,28 @@
+# noinspection PyUnresolvedReferences
+from django.conf import settings
+from django.core.management.base import CommandError
+from django.core.management.commands.makemigrations import Command as _Command
+from django.db import models
+
+from utilities.migration import custom_deconstruct
+
+models.Field.deconstruct = custom_deconstruct
+
+
+class Command(_Command):
+
+    def handle(self, *args, **kwargs):
+        """
+        This built-in management command enables the creation of new database schema migration files, which should
+        never be required by and ordinary user. We prevent this command from executing unless the configuration
+        indicates that the user is a developer (i.e. configuration.DEVELOPER == True).
+        """
+        if not settings.DEVELOPER:
+            raise CommandError(
+                "This command is available for development purposes only. It will\n"
+                "NOT resolve any issues with missing or unapplied migrations. For assistance,\n"
+                "please post to the NetBox discussion forum on GitHub:\n"
+                "    https://github.com/netbox-community/netbox/discussions"
+            )
+
+        super().handle(*args, **kwargs)

netbox/core/management/commands/migrate.py

@@ -0,0 +1,7 @@
+# noinspection PyUnresolvedReferences
+from django.core.management.commands.migrate import Command
+from django.db import models
+
+from utilities.migration import custom_deconstruct
+
+models.Field.deconstruct = custom_deconstruct

netbox/core/management/commands/nbshell.py

@@ -0,0 +1,92 @@
+import code
+import platform
+import sys
+
+from django import get_version
+from django.apps import apps
+from django.conf import settings
+from django.contrib.auth.models import User
+from django.contrib.contenttypes.models import ContentType
+from django.core.management.base import BaseCommand
+
+APPS = ('circuits', 'core', 'dcim', 'extras', 'ipam', 'tenancy', 'users', 'virtualization', 'wireless')
+
+BANNER_TEXT = """### NetBox interactive shell ({node})
+### Python {python} | Django {django} | NetBox {netbox}
+### lsmodels() will show available models. Use help(<model>) for more info.""".format(
+    node=platform.node(),
+    python=platform.python_version(),
+    django=get_version(),
+    netbox=settings.VERSION
+)
+
+
+class Command(BaseCommand):
+    help = "Start the Django shell with all NetBox models already imported"
+    django_models = {}
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '-c', '--command',
+            help='Python code to execute (instead of starting an interactive shell)',
+        )
+
+    def _lsmodels(self):
+        for app, models in self.django_models.items():
+            app_name = apps.get_app_config(app).verbose_name
+            print(f'{app_name}:')
+            for m in models:
+                print(f'  {m}')
+
+    def get_namespace(self):
+        namespace = {}
+
+        # Gather Django models and constants from each app
+        for app in APPS:
+            self.django_models[app] = []
+
+            # Load models from each app
+            for model in apps.get_app_config(app).get_models():
+                namespace[model.__name__] = model
+                self.django_models[app].append(model.__name__)
+
+            # Constants
+            try:
+                app_constants = sys.modules[f'{app}.constants']
+                for name in dir(app_constants):
+                    namespace[name] = getattr(app_constants, name)
+            except KeyError:
+                pass
+
+        # Additional objects to include
+        namespace['ContentType'] = ContentType
+        namespace['User'] = User
+
+        # Load convenience commands
+        namespace.update({
+            'lsmodels': self._lsmodels,
+        })
+
+        return namespace
+
+    def handle(self, **options):
+        namespace = self.get_namespace()
+
+        # If Python code has been passed, execute it and exit.
+        if options['command']:
+            exec(options['command'], namespace)
+            return
+
+        # Try to enable tab-complete
+        try:
+            import readline
+            import rlcompleter
+        except ModuleNotFoundError:
+            pass
+        else:
+            readline.set_completer(rlcompleter.Completer(namespace).complete)
+            readline.parse_and_bind('tab: complete')
+
+        # Run interactive shell
+        shell = code.interact(banner=BANNER_TEXT, local=namespace)
+        return shell

netbox/core/management/commands/rqworker.py

@@ -0,0 +1,28 @@
+import logging
+
+from django_rq.management.commands.rqworker import Command as _Command
+
+
+DEFAULT_QUEUES = ('high', 'default', 'low')
+
+logger = logging.getLogger('netbox.rqworker')
+
+
+class Command(_Command):
+    """
+    Subclass django_rq's built-in rqworker to listen on all configured queues if none are specified (instead
+    of only the 'default' queue).
+    """
+    def handle(self, *args, **options):
+        # Run the worker with scheduler functionality
+        options['with_scheduler'] = True
+
+        # If no queues have been specified on the command line, listen on all configured queues.
+        if len(args) < 1:
+            queues = ', '.join(DEFAULT_QUEUES)
+            logger.warning(
+                f"No queues have been specified. This process will service the following queues by default: {queues}"
+            )
+            args = DEFAULT_QUEUES
+
+        super().handle(*args, **options)