Closes #22772: Allow dynamic registration of background job result URLs

docs/development/application-registry.md

@@ -20,6 +20,10 @@ A dictionary mapping data backend types to their respective classes. These are u
 
 A dictionary mapping each model (identified by its app and label) to its filterset class, if one has been registered for it. Filtersets are registered using the `@register_filterset` decorator.
 
+### `job_result_urls`
+
+A dictionary mapping each model (identified by its app and label) and optional job name to a function which returns the URL of the results page for background jobs associated with objects of that model. These functions are registered using the `register_job_result_url()` decorator in `core.utils`. (See the [plugin development documentation](../plugins/development/background-jobs.md#result-urls) for details.)
+
 ### `model_features`
 
 A dictionary of model features (e.g. custom fields, tags, etc.) mapped to the functions used to qualify a model as supporting each feature. Model features are registered using the `register_model_feature()` function in `netbox.utils`.

docs/plugins/development/background-jobs.md

@@ -93,6 +93,27 @@ class MyModel(JobsMixin, NetBoxModel):
 ```
 
 
+### Result URLs
+
+By default, links to a job (such as those in the jobs list and in job completion notifications) lead to its generic view. A plugin which provides its own page for presenting the results of jobs associated with one of its models can register a function which returns the URL of that page, using the `register_job_result_url()` decorator from `core.utils`. The function accepts a Job instance and returns either a URL, or `None` to fall back to the generic view.
+
+Optionally, a job name may be specified to restrict the function to jobs having that name. A function registered for a specific job name takes precedence over one registered for the model alone. Only one function may be registered for each combination of model and job name.
+
+#### Example
+
+```python title="jobs.py"
+from django.urls import reverse
+from core.utils import register_job_result_url
+from .models import MyModel
+
+@register_job_result_url(MyModel, job_name='My Test Job')
+def my_test_job_result_url(job):
+    return reverse('plugins:my_plugin:mymodel_results', args=[job.object_id])
+```
+
+!!! note
+    As with [system jobs](#system-jobs), ensure that any such functions are imported on initialization. Otherwise, they won't be registered.
+
 ### System Jobs
 
 Some plugins may implement background jobs that are decoupled from the request/response cycle. Typical use cases would be housekeeping tasks or synchronization jobs. These can be registered as _system jobs_ using the `system_job()` decorator. The job interval must be passed as an integer (in minutes) when registering a system job. System jobs are scheduled automatically when the RQ worker (`manage.py rqworker`) is run.

netbox/core/models/jobs.py

@@ -21,6 +21,7 @@ from core.dataclasses import JobLogEntry
 from core.events import JOB_COMPLETED, JOB_ERRORED, JOB_FAILED
 from core.models import ObjectType
 from core.signals import job_end, job_start
+from core.utils import get_job_result_url
 from extras.models import Notification
 from netbox.models.features import has_feature
 from utilities.json import JobLogDecoder
@@ -155,12 +156,8 @@ class Job(models.Model):
         return self.name
 
     def get_absolute_url(self):
-        # TODO: Employ dynamic registration
-        if self.object_type:
-            if self.object_type.model == 'reportmodule':
-                return reverse('extras:report_result', kwargs={'job_pk': self.pk})
-            if self.object_type.model == 'scriptmodule':
-                return reverse('extras:script_result', kwargs={'job_pk': self.pk})
+        if url := get_job_result_url(self):
+            return url
         return reverse('core:job', args=[self.pk])
 
     def get_status_color(self):

netbox/core/tests/test_models.py

@@ -12,6 +12,7 @@ from django.core.management import call_command
 from django.db import IntegrityError, connection
 from django.test import TestCase, override_settings
 from django.test.utils import CaptureQueriesContext
+from django.urls import reverse
 from django.utils import timezone
 
 from core.choices import DataSourceStatusChoices, JobNotificationChoices, JobStatusChoices, ObjectChangeActionChoices
@@ -20,7 +21,7 @@ from core.exceptions import SyncError
 from core.models import AutoSyncRecord, DataFile, DataSource, Job, ObjectType
 from core.signals import post_sync
 from dcim.models import Device, Location, Site
-from extras.models import ConfigTemplate, Notification
+from extras.models import ConfigTemplate, Notification, Script
 from netbox.constants import CENSOR_TOKEN, CENSOR_TOKEN_CHANGED
 from users.models import User
 from utilities.querysets import RestrictedQuerySet
@@ -505,3 +506,12 @@ class JobTestCase(TestCase):
         job.terminate(status=JobStatusChoices.STATUS_COMPLETED)
 
         self.assertIsNone(job.execution_time)
+
+    def test_get_absolute_url(self):
+        job = Job(pk=123, name='Test Job')
+        self.assertEqual(job.get_absolute_url(), reverse('core:job', args=[123]))
+
+        # Jobs for custom scripts link to the script result view
+        job.object_type = ObjectType.objects.get_for_model(Script)
+        job.object_id = 1
+        self.assertEqual(job.get_absolute_url(), reverse('extras:script_result', kwargs={'job_pk': 123}))

netbox/core/tests/test_utils.py

@@ -0,0 +1,86 @@
+from unittest.mock import patch
+
+from django.contrib.contenttypes.models import ContentType
+from django.core.exceptions import ImproperlyConfigured
+from django.test import TestCase
+from django.urls import reverse
+
+from core.models import DataSource, Job, ObjectType
+from core.utils import register_job_result_url
+from dcim.models import Site
+from netbox.registry import registry
+
+
+class JobResultURLTestCase(TestCase):
+
+    def setUp(self):
+        patcher = patch.dict(registry, {'job_result_urls': {}})
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    @staticmethod
+    def get_job(model=None, name='Test Job'):
+        object_type = ObjectType.objects.get_for_model(model) if model else None
+        return Job(pk=123, name=name, object_type=object_type, object_id=1 if model else None)
+
+    def test_register_job_result_url(self):
+        def func(job):
+            return None
+
+        self.assertIs(register_job_result_url(DataSource)(func), func)
+        self.assertIs(register_job_result_url(DataSource, job_name='Sync')(func), func)
+        self.assertEqual(registry['job_result_urls'], {
+            ('core.datasource', None): func,
+            ('core.datasource', 'Sync'): func,
+        })
+
+    def test_register_duplicate(self):
+        register_job_result_url(DataSource, job_name='Sync')(lambda job: None)
+        with self.assertRaises(ImproperlyConfigured):
+            register_job_result_url(DataSource, job_name='Sync')(lambda job: None)
+
+    def test_no_object(self):
+        register_job_result_url(DataSource)(lambda job: '/foo/')
+
+        self.assertEqual(self.get_job().get_absolute_url(), reverse('core:job', args=[123]))
+
+    def test_no_registration_for_model(self):
+        register_job_result_url(DataSource)(lambda job: '/foo/')
+
+        self.assertEqual(self.get_job(Site).get_absolute_url(), reverse('core:job', args=[123]))
+
+    def test_model(self):
+        register_job_result_url(DataSource)(lambda job: f'/foo/{job.object_id}/')
+
+        self.assertEqual(self.get_job(DataSource).get_absolute_url(), '/foo/1/')
+
+    def test_job_name_takes_precedence(self):
+        register_job_result_url(DataSource)(lambda job: '/foo/')
+        register_job_result_url(DataSource, job_name='Sync')(lambda job: '/bar/')
+
+        self.assertEqual(self.get_job(DataSource, name='Sync').get_absolute_url(), '/bar/')
+        self.assertEqual(self.get_job(DataSource, name='Other').get_absolute_url(), '/foo/')
+
+    def test_declined(self):
+        register_job_result_url(DataSource, job_name='Sync')(lambda job: None)
+
+        self.assertEqual(self.get_job(DataSource, name='Sync').get_absolute_url(), reverse('core:job', args=[123]))
+
+        # Declining defers to the function registered for the model alone
+        register_job_result_url(DataSource)(lambda job: '/foo/')
+        self.assertEqual(self.get_job(DataSource, name='Sync').get_absolute_url(), '/foo/')
+
+    def test_exception(self):
+        @register_job_result_url(DataSource)
+        def func(job):
+            return reverse('plugins:invalid')
+
+        with self.assertLogs('netbox.jobs', level='ERROR'):
+            self.assertEqual(self.get_job(DataSource).get_absolute_url(), reverse('core:job', args=[123]))
+
+    def test_unresolvable_model(self):
+        # e.g. a model belonging to a plugin which has been uninstalled
+        object_type = ContentType.objects.create(app_label='uninstalled_plugin', model='foo')
+        job = Job(pk=123, name='Test Job', object_type=object_type, object_id=1)
+
+        self.assertEqual(job.get_absolute_url(), reverse('core:job', args=[123]))

netbox/core/utils.py

@@ -1,3 +1,6 @@
+import logging
+
+from django.core.exceptions import ImproperlyConfigured
 from django.db import DatabaseError, connection
 from django.http import Http404
 from django.utils.translation import gettext_lazy as _
@@ -16,12 +19,16 @@ from rq.registry import (
     StartedJobRegistry,
 )
 
+from netbox.registry import registry
+
 __all__ = (
     'delete_rq_job',
     'enqueue_rq_job',
     'get_db_schema',
+    'get_job_result_url',
     'get_rq_jobs',
     'get_rq_jobs_from_status',
+    'register_job_result_url',
     'requeue_rq_job',
     'stop_rq_job',
 )
@@ -203,3 +210,49 @@ def get_db_schema():
     except DatabaseError:
         pass
     return db_schema
+
+
+def register_job_result_url(model, job_name=None):
+    """
+    Decorator for registering a function which returns the URL of the results page for background jobs associated
+    with objects of the given model (and optionally, having the given name).
+
+    The function must accept a Job instance and return either a URL, or None to defer to the next applicable function
+    (and ultimately to the generic job view). A function registered for a specific job name takes precedence over one
+    registered for the model alone.
+
+        @register_job_result_url(MyModel, job_name='My Job')
+        def my_job_result_url(job):
+            return reverse('plugins:myplugin:mymodel_results', args=[job.object_id])
+    """
+    key = (model._meta.label_lower, job_name)
+
+    def _wrapper(func):
+        if key in registry['job_result_urls']:
+            raise ImproperlyConfigured(
+                f"A job result URL function is already registered for model {model._meta.label} and job name "
+                f"{job_name!r}."
+            )
+        registry['job_result_urls'][key] = func
+        return func
+
+    return _wrapper
+
+
+def get_job_result_url(job):
+    """
+    Return the URL provided by the first applicable job result URL function registered for the given Job, if any.
+    """
+    if not job.object_type:
+        return None
+    model = f'{job.object_type.app_label}.{job.object_type.model}'
+    for key in ((model, job.name), (model, None)):
+        if func := registry['job_result_urls'].get(key):
+            try:
+                if url := func(job):
+                    return url
+            except Exception:
+                # A failing function must not break the rendering of jobs lists or notifications.
+                logger = logging.getLogger('netbox.jobs')
+                logger.exception("Failed to resolve the result URL for job %s using %r", job.pk, func)
+    return None

netbox/extras/apps.py

@@ -8,7 +8,7 @@ class ExtrasConfig(AppConfig):
         from netbox.event_rules import register_event_rule_action
         from netbox.models.features import register_models
 
-        from . import dashboard, lookups, search, signals  # noqa: F401
+        from . import dashboard, jobs, lookups, search, signals  # noqa: F401
         from .event_rules import NotificationAction, ScriptAction, WebhookAction
 
         # Register models

netbox/extras/jobs.py

@@ -4,10 +4,12 @@ from contextlib import ExitStack
 
 from django.apps import apps
 from django.db import DEFAULT_DB_ALIAS, router, transaction
+from django.urls import reverse
 from django.utils.translation import gettext as _
 from django_pg_utils import advisory_lock
 
 from core.signals import clear_events
+from core.utils import register_job_result_url
 from dcim.models import Device
 from extras.choices import CustomFieldStatusChoices
 from extras.constants import CUSTOMFIELD_JOB_TIMEOUT
@@ -260,6 +262,11 @@ class CustomFieldPurgeJob(CustomFieldDataJob):
 #
 
 
+@register_job_result_url(ScriptModel)
+def script_job_result_url(job):
+    return reverse('extras:script_result', kwargs={'job_pk': job.pk})
+
+
 class ScriptJob(JobRunner):
     """
     Script execution job.

netbox/netbox/registry.py

@@ -28,6 +28,7 @@ registry = Registry({
     'event_rule_actions': dict(),
     'event_types': dict(),
     'filtersets': dict(),
+    'job_result_urls': dict(),
     'model_actions': collections.defaultdict(set),
     'model_features': dict(),
     'pk_allocators': list(),