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

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 `(model_label, job_name)` pairs to `(function, raise_exceptions)` tuples registered with `core.utils.register_job_result_url()`. Model labels use `<app_label>.<model_name>`; a `None` job name provides a fallback for that model. See [Result URLs](../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,31 @@ class MyModel(JobsMixin, NetBoxModel):
 ```
 
 
+### Result URLs
+
+Use `core.utils.register_job_result_url()` to link jobs to your plugin's result page. Pass a model class or an `<app_label>.<model_name>` string, and optionally a `job_name`. The function receives a `Job` and returns a URL, or `None` to try the next fallback.
+
+NetBox tries the registration for the model and job name first, then the model alone, then the generic Job view. Each model/name pair can be registered only once.
+
+`NoReverseMatch` and `ObjectDoesNotExist` are logged as warnings; any other exception is logged as an error with its traceback. Either way, the function is treated as having returned `None`. If your function uses `job.object`, handle it being `None` after deletion.
+
+The URL is cached on each `Job` instance. Later changes to that instance do not update the cached URL.
+
+#### 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

@@ -1,7 +1,7 @@
 import logging
 import uuid
 from dataclasses import asdict
-from functools import partial
+from functools import cached_property, partial
 
 import django_rq
 from django.conf import settings
@@ -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,13 +156,12 @@ 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})
-        return reverse('core:job', args=[self.pk])
+        return self._absolute_url
+
+    @cached_property
+    def _absolute_url(self):
+        # Cached because job tables link multiple columns, and a registered result URL function may query the database
+        return get_job_result_url(self) or reverse('core:job', args=[self.pk])
 
     def get_status_color(self):
         return JobStatusChoices.colors.get(self.status)

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,11 @@ 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 = Job(pk=123, name='Test Job', object_type=ObjectType.objects.get_for_model(Script), 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,126 @@
+from unittest.mock import Mock, patch
+
+from django.contrib.contenttypes.models import ContentType
+from django.core.exceptions import ImproperlyConfigured
+from django.test import TestCase
+from django.urls import NoReverseMatch, 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.assertIs(register_job_result_url('dcim.Site', raise_exceptions=True)(func), func)
+        self.assertEqual(registry['job_result_urls'], {
+            ('core.datasource', None): (func, False),
+            ('core.datasource', 'Sync'): (func, False),
+            ('dcim.site', None): (func, True),
+        })
+
+    def test_register_duplicate(self):
+        register_job_result_url(DataSource, job_name='Sync')(lambda job: None)
+        with self.assertRaises(ImproperlyConfigured):
+            register_job_result_url('core.datasource', job_name='Sync')(lambda job: None)
+
+    def test_register_invalid_label(self):
+        for label in ('datasource', 'core.datasource.foo', 'core.', '.datasource', '.'):
+            with self.subTest(label=label), self.assertRaises(ImproperlyConfigured):
+                register_job_result_url(label)
+
+    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='WARNING'):
+            self.assertEqual(self.get_job(DataSource).get_absolute_url(), reverse('core:job', args=[123]))
+
+    def test_unexpected_exception(self):
+        @register_job_result_url(DataSource, job_name='Sync')
+        def func(job):
+            raise AttributeError
+
+        with self.assertLogs('netbox.jobs', level='ERROR'):
+            self.assertEqual(self.get_job(DataSource, name='Sync').get_absolute_url(), reverse('core:job', args=[123]))
+
+        # A failing function defers to the function registered for the model alone
+        register_job_result_url(DataSource)(lambda job: '/foo/')
+        with self.assertLogs('netbox.jobs', level='ERROR'):
+            self.assertEqual(self.get_job(DataSource, name='Sync').get_absolute_url(), '/foo/')
+
+    def test_raise_exceptions(self):
+        func = Mock()
+        register_job_result_url(DataSource, raise_exceptions=True)(func)
+
+        for exception in (NoReverseMatch, AttributeError):
+            func.side_effect = exception
+            with self.subTest(exception=exception.__name__), self.assertRaises(exception):
+                self.get_job(DataSource).get_absolute_url()
+
+    def test_url_cached(self):
+        func = Mock(return_value='/foo/')
+        register_job_result_url(DataSource)(func)
+        job = self.get_job(DataSource)
+
+        self.assertEqual(job.get_absolute_url(), '/foo/')
+        self.assertEqual(job.get_absolute_url(), '/foo/')
+        func.assert_called_once_with(job)
+
+    def test_unresolvable_model(self):
+        register_job_result_url(DataSource)(lambda job: '/foo/')
+
+        # 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,5 +1,9 @@
+import logging
+
+from django.core.exceptions import ImproperlyConfigured, ObjectDoesNotExist
 from django.db import DatabaseError, connection
 from django.http import Http404
+from django.urls import NoReverseMatch
 from django.utils.translation import gettext_lazy as _
 from django_rq.queues import get_queue, get_queue_by_index, get_redis_connection
 from django_rq.settings import get_queues_list, get_queues_map
@@ -16,16 +20,22 @@ 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',
 )
 
+logger = logging.getLogger('netbox.jobs')
+
 
 def get_rq_jobs():
     """
@@ -203,3 +213,67 @@ def get_db_schema():
     except DatabaseError:
         pass
     return db_schema
+
+
+def register_job_result_url(model, job_name=None, *, raise_exceptions=False):
+    """
+    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 model may be passed as a class or as
+    a string in the form "<app_label>.<model_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])
+
+    If raise_exceptions is True, exceptions raised by the function propagate instead of being logged. NetBox's own
+    registrations set it, so that core bugs aren't masked.
+    """
+    if isinstance(model, str):
+        if len(parts := model.split('.')) != 2 or not all(parts):
+            raise ImproperlyConfigured(f"Invalid model label {model!r}: must be in the form <app_label>.<model_name>.")
+        label = model.lower()
+    else:
+        label = model._meta.label_lower
+    key = (label, 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 {label} and job name {job_name!r}."
+            )
+        registry['job_result_urls'][key] = (func, raise_exceptions)
+        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 not (registration := registry['job_result_urls'].get(key)):
+            continue
+        func, raise_exceptions = registration
+        try:
+            if url := func(job):
+                return url
+        except Exception as e:
+            # A failing function must not break the rendering of jobs lists or notifications.
+            if raise_exceptions:
+                raise
+            if isinstance(e, (NoReverseMatch, ObjectDoesNotExist)):
+                logger.warning(
+                    "Failed to resolve the result URL for job %s using %r: %s", job.pk, func, e,
+                    exc_info=logger.isEnabledFor(logging.DEBUG)
+                )
+            else:
+                logger.exception("Error resolving 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, raise_exceptions=True)
+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

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