Per-Script Worker Queue Setting

docs/customization/custom-scripts.md

@@ -108,6 +108,10 @@ commit_default = False
 
 By default, a script can be scheduled for execution at a later time. Setting `scheduling_enabled` to False disables this ability: Only immediate execution will be possible. (This also disables the ability to set a recurring execution interval.)
 
+### `rq_queue_name`
+
+This will override the standard `QUEUE_MAPPINGS` setting for the script to be process by the worker.  If you use a queue name that is not one of the NetBox provided queues, (IE `high`, `default`, or `low`) then you must reconfigure the RQ working for the new queue. If the queue is not configured then the default logic found in the settings file for [QUEUE_MAPPINGS](../configuration/miscellaneous.md#queue_mappings) will be used.
+
 ### `job_timeout`
 
 Set the maximum allowed runtime for the script. If not set, `RQ_DEFAULT_TIMEOUT` will be used.

netbox/core/models/jobs.py

@@ -1,6 +1,8 @@
+import logging
 import uuid
 
 import django_rq
+from rq.exceptions import NoSuchJobError
 from django.conf import settings
 from django.contrib.contenttypes.fields import GenericForeignKey
 from django.core.exceptions import ValidationError
@@ -23,6 +25,8 @@
     'Job',
 )
 
+logger = logging.getLogger('netbox.core.jobs')
+
 
 class Job(models.Model):
     """
@@ -152,10 +156,14 @@ def duration(self):
 
     def delete(self, *args, **kwargs):
         super().delete(*args, **kwargs)
-
-        rq_queue_name = get_config().QUEUE_MAPPINGS.get(self.object_type.model, RQ_QUEUE_DEFAULT)
-        queue = django_rq.get_queue(rq_queue_name)
-        job = queue.fetch_job(str(self.job_id))
+        job = None
+        for check_queue in get_config().QUEUE_MAPPINGS:
+            queue = django_rq.get_queue(check_queue)
+            try:
+                job = queue.fetch_job(str(self.job_id))
+                break
+            except NoSuchJobError:
+                pass
 
         if job:
             job.cancel()
@@ -198,7 +206,7 @@ def terminate(self, status=JobStatusChoices.STATUS_COMPLETED, error=None):
         job_end.send(self)
 
     @classmethod
-    def enqueue(cls, func, instance, name='', user=None, schedule_at=None, interval=None, **kwargs):
+    def enqueue(cls, func, instance, name='', user=None, schedule_at=None, interval=None, rq_queue_name=None, **kwargs):
         """
         Create a Job instance and enqueue a job using the given callable
 
@@ -208,11 +216,19 @@ def enqueue(cls, func, instance, name='', user=None, schedule_at=None, interval=
             name: Name for the job (optional)
             user: The user responsible for running the job
             schedule_at: Schedule the job to be executed at the passed date and time
+            rq_queue_name: Queue name to route the job to for processing
             interval: Recurrence interval (in minutes)
         """
         object_type = ObjectType.objects.get_for_model(instance, for_concrete_model=False)
-        rq_queue_name = get_queue_for_model(object_type.model)
-        queue = django_rq.get_queue(rq_queue_name)
+        queue = None
+        if rq_queue_name:
+            try:
+                queue = django_rq.get_queue(rq_queue_name)
+            except Exception:
+                logger.warning(f"User defined queue '{rq_queue_name}' cased an error or was not found. Falling back to default queue.")
+        if not queue:
+            rq_queue_name = get_queue_for_model(object_type.model)
+            queue = django_rq.get_queue(rq_queue_name)
         status = JobStatusChoices.STATUS_SCHEDULED if schedule_at else JobStatusChoices.STATUS_PENDING
         job = Job.objects.create(
             object_type=object_type,

netbox/extras/api/views.py

@@ -261,6 +261,7 @@ def post(self, request, pk):
                 request=copy_safe_request(request),
                 commit=input_serializer.data['commit'],
                 job_timeout=script.python_class.job_timeout,
+                rq_queue_name=script.python_class.rq_queue_name,
                 schedule_at=input_serializer.validated_data.get('schedule_at'),
                 interval=input_serializer.validated_data.get('interval')
             )

netbox/extras/scripts.py

@@ -373,6 +373,10 @@ def job_timeout(self):
     def scheduling_enabled(self):
         return getattr(self.Meta, 'scheduling_enabled', True)
 
+    @classproperty
+    def rq_queue_name(self):
+        return getattr(self.Meta, 'rq_queue_name', None)
+
     @property
     def filename(self):
         return inspect.getfile(self.__class__)
@@ -717,6 +721,7 @@ def _run_script(job):
             schedule_at=new_scheduled_time,
             interval=job.interval,
             job_timeout=script.job_timeout,
+            rq_queue_name=script.rq_queue_name,
             data=data,
             request=request,
             commit=commit

netbox/extras/views.py

@@ -1110,6 +1110,7 @@ def post(self, request, **kwargs):
                 data=form.cleaned_data,
                 request=copy_safe_request(request),
                 job_timeout=script.python_class.job_timeout,
+                rq_queue_name=script.python_class.rq_queue_name,
                 commit=form.cleaned_data.pop('_commit')
             )