Reference ObjectType records instead of registry for feature support

netbox/extras/events.py

@@ -9,9 +9,9 @@ from django.utils.translation import gettext as _
 from django_rq import get_queue
 
 from core.events import *
+from core.models import ObjectType
 from netbox.config import get_config
 from netbox.constants import RQ_QUEUE_DEFAULT
-from netbox.registry import registry
 from users.models import User
 from utilities.api import get_serializer_for_model
 from utilities.rqworker import get_rq_retry
@@ -55,11 +55,12 @@ def enqueue_event(queue, instance, user, request_id, event_type):
     Enqueue a serialized representation of a created/updated/deleted object for the processing of
     events once the request has completed.
     """
-    # Determine whether this type of object supports event rules
+    # Bail if this type of object does not support event rules
+    if 'event_rules' not in ObjectType.objects.get_for_model(instance).features:
+        return
+
     app_label = instance._meta.app_label
     model_name = instance._meta.model_name
-    if model_name not in registry['model_features']['event_rules'].get(app_label, []):
-        return
 
     assert instance.pk is not None
     key = f'{app_label}.{model_name}:{instance.pk}'

netbox/extras/models/configs.py

@@ -1,15 +1,16 @@
-from django.apps import apps
+from collections import defaultdict
+
 from django.conf import settings
 from django.core.validators import ValidationError
 from django.db import models
 from django.urls import reverse
 from django.utils.translation import gettext_lazy as _
 
+from core.models import ObjectType
 from extras.models.mixins import RenderTemplateMixin
 from extras.querysets import ConfigContextQuerySet
 from netbox.models import ChangeLoggedModel
 from netbox.models.features import CloningMixin, CustomLinksMixin, ExportTemplatesMixin, SyncedDataMixin, TagsMixin
-from netbox.registry import registry
 from utilities.data import deepmerge
 
 __all__ = (
@@ -239,15 +240,12 @@ class ConfigTemplate(
     sync_data.alters_data = True
 
     def get_context(self, context=None, queryset=None):
-        _context = dict()
-        for app, model_names in registry['models'].items():
-            _context.setdefault(app, {})
-            for model_name in model_names:
-                try:
-                    model = apps.get_registered_model(app, model_name)
-                    _context[app][model.__name__] = model
-                except LookupError:
-                    pass
+        _context = defaultdict(dict)
+
+        # Populate all public models for reference within the template
+        for object_type in ObjectType.objects.public():
+            if model := object_type.model_class():
+                _context[object_type.app_label][model.__name__] = model
 
         # Apply the provided context data, if any
         if context is not None:

netbox/extras/signals.py

@@ -8,7 +8,6 @@ from core.signals import job_end, job_start
 from extras.events import process_event_rules
 from extras.models import EventRule, Notification, Subscription
 from netbox.config import get_config
-from netbox.registry import registry
 from netbox.signals import post_clean
 from utilities.exceptions import AbortRequest
 from .models import CustomField, TaggedItem
@@ -150,17 +149,24 @@ def notify_object_changed(sender, instance, **kwargs):
         event_type = OBJECT_DELETED
 
     # Skip unsupported object types
-    ct = ContentType.objects.get_for_model(instance)
-    if ct.model not in registry['model_features']['notifications'].get(ct.app_label, []):
+    object_type = ObjectType.objects.get_for_model(instance)
+    if 'notifications' not in object_type.features:
         return
 
     # Find all subscribed Users
-    subscribed_users = Subscription.objects.filter(object_type=ct, object_id=instance.pk).values_list('user', flat=True)
+    subscribed_users = Subscription.objects.filter(
+        object_type=object_type,
+        object_id=instance.pk
+    ).values_list('user', flat=True)
     if not subscribed_users:
         return
 
     # Delete any existing Notifications for the object
-    Notification.objects.filter(object_type=ct, object_id=instance.pk, user__in=subscribed_users).delete()
+    Notification.objects.filter(
+        object_type=object_type,
+        object_id=instance.pk,
+        user__in=subscribed_users
+    ).delete()
 
     # Create Notifications for Subscribers
     Notification.objects.bulk_create([

netbox/netbox/models/features.py

@@ -32,6 +32,7 @@ __all__ = (
     'CustomValidationMixin',
     'EventRulesMixin',
     'ExportTemplatesMixin',
+    'FEATURES_MAP',
     'ImageAttachmentsMixin',
     'JobsMixin',
     'JournalingMixin',
@@ -633,6 +634,7 @@ FEATURES_MAP = {
     'tags': TagsMixin,
 }
 
+# TODO: Remove in NetBox v4.5
 registry['model_features'].update({
     feature: defaultdict(set) for feature in FEATURES_MAP.keys()
 })
@@ -653,10 +655,12 @@ def register_models(*models):
     for model in models:
         app_label, model_name = model._meta.label_lower.split('.')
 
+        # TODO: Remove in NetBox v4.5
         # Register public models
         if not getattr(model, '_netbox_private', False):
             registry['models'][app_label].add(model_name)
 
+        # TODO: Remove in NetBox v4.5
         # Record each applicable feature for the model in the registry
         features = {
             feature for feature, cls in FEATURES_MAP.items() if issubclass(model, cls)

netbox/netbox/tests/test_plugins.py

@@ -6,6 +6,7 @@ from django.test import Client, TestCase, override_settings
 from django.urls import reverse
 
 from core.choices import JobIntervalChoices
+from core.models import ObjectType
 from netbox.tests.dummy_plugin import config as dummy_config
 from netbox.tests.dummy_plugin.data_backends import DummyBackend
 from netbox.tests.dummy_plugin.jobs import DummySystemJob
@@ -23,8 +24,9 @@ class PluginTest(TestCase):
         self.assertIn('netbox.tests.dummy_plugin.DummyPluginConfig', settings.INSTALLED_APPS)
 
     def test_model_registration(self):
-        self.assertIn('dummy_plugin', registry['models'])
-        self.assertIn('dummymodel', registry['models']['dummy_plugin'])
+        self.assertIsNone(
+            ObjectType.objects.filter(app_label='dummy_plugin', model='dummymodel')
+        )
 
     def test_models(self):
         from netbox.tests.dummy_plugin.models import DummyModel