This commit is contained in:
Jeremy Stretch 2025-07-24 10:26:26 -04:00 committed by GitHub
commit 4509c52579
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 343 additions and 65 deletions

View File

@ -0,0 +1,55 @@
import django.contrib.postgres.fields
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('contenttypes', '0002_remove_content_type_name'),
('core', '0016_job_log_entries'),
]
operations = [
# Delete the proxy model from the migration state
migrations.DeleteModel(
name='ObjectType',
),
# Create the new concrete model
migrations.CreateModel(
name='ObjectType',
fields=[
(
'contenttype_ptr',
models.OneToOneField(
on_delete=django.db.models.deletion.CASCADE,
parent_link=True,
primary_key=True,
serialize=False,
to='contenttypes.contenttype',
related_name='object_type'
)
),
(
'public',
models.BooleanField(
default=False
)
),
(
'features',
django.contrib.postgres.fields.ArrayField(
base_field=models.CharField(max_length=50),
default=list,
size=None
)
),
],
options={
'verbose_name': 'object type',
'verbose_name_plural': 'object types',
},
bases=('contenttypes.contenttype',),
managers=[],
),
]

View File

@ -11,8 +11,8 @@ from mptt.models import MPTTModel
from core.choices import ObjectChangeActionChoices
from core.querysets import ObjectChangeQuerySet
from netbox.models.features import ChangeLoggingMixin
from netbox.models.features import has_feature
from utilities.data import shallow_compare_dict
from .contenttypes import ObjectType
__all__ = (
'ObjectChange',
@ -118,7 +118,7 @@ class ObjectChange(models.Model):
super().clean()
# Validate the assigned object type
if self.changed_object_type not in ObjectType.objects.with_feature('change_logging'):
if not has_feature(self.changed_object_type, 'change_logging'):
raise ValidationError(
_("Change logging is not supported for this object type ({type}).").format(
type=self.changed_object_type

View File

@ -1,5 +1,8 @@
from django.contrib.contenttypes.models import ContentType, ContentTypeManager
from django.db.models import Q
from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.fields import ArrayField
from django.core.exceptions import ObjectDoesNotExist
from django.db import models
from django.utils.translation import gettext as _
from netbox.plugins import PluginConfig
from netbox.registry import registry
@ -11,45 +14,117 @@ __all__ = (
)
class ObjectTypeManager(ContentTypeManager):
class ObjectTypeQuerySet(models.QuerySet):
def create(self, **kwargs):
# If attempting to create a new ObjectType for a given app_label & model, replace those kwargs
# with a reference to the ContentType (if one exists).
if (app_label := kwargs.get('app_label')) and (model := kwargs.get('model')):
try:
kwargs['contenttype_ptr'] = ContentType.objects.get(app_label=app_label, model=model)
except ObjectDoesNotExist:
pass
return super().create(**kwargs)
class ObjectTypeManager(models.Manager):
def get_queryset(self):
return ObjectTypeQuerySet(self.model, using=self._db)
def get_by_natural_key(self, app_label, model):
"""
Retrieve an ObjectType by its application label & model name.
This method exists to provide parity with ContentTypeManager.
"""
return self.get(app_label=app_label, model=model)
# TODO: Remove in NetBox v4.5
def get_for_id(self, id):
"""
Retrieve an ObjectType by its primary key (numeric ID).
This method exists to provide parity with ContentTypeManager.
"""
return self.get(pk=id)
def _get_opts(self, model, for_concrete_model):
if for_concrete_model:
model = model._meta.concrete_model
return model._meta
def get_for_model(self, model, for_concrete_model=True):
"""
Retrieve or create and return the ObjectType for a model.
"""
from netbox.models.features import get_model_features, model_is_public
opts = self._get_opts(model, for_concrete_model)
try:
# Use .get() instead of .get_or_create() initially to ensure db_for_read is honored (Django bug #20401).
ot = self.get(app_label=opts.app_label, model=opts.model_name)
except self.model.DoesNotExist:
# If the ObjectType doesn't exist, create it. (Use .get_or_create() to avoid race conditions.)
ot = self.get_or_create(
app_label=opts.app_label,
model=opts.model_name,
public=model_is_public(model),
features=get_model_features(model.__class__),
)[0]
return ot
def public(self):
"""
Filter the base queryset to return only ContentTypes corresponding to "public" models; those which are listed
in registry['models'] and intended for reference by other objects.
Includes only ObjectTypes for "public" models.
Filter the base queryset to return only ObjectTypes corresponding to public models; those which are intended
for reference by other objects within the application.
"""
q = Q()
for app_label, models in registry['models'].items():
q |= Q(app_label=app_label, model__in=models)
return self.get_queryset().filter(q)
return self.get_queryset().filter(public=True)
def with_feature(self, feature):
"""
Return the ContentTypes only for models which are registered as supporting the specified feature. For example,
we can find all ContentTypes for models which support webhooks with
Return ObjectTypes only for models which support the given feature.
ContentType.objects.with_feature('event_rules')
Only ObjectTypes which list the specified feature will be included. Supported features are declared in
netbox.models.features.FEATURES_MAP. For example, we can find all ObjectTypes for models which support event
rules with:
ObjectType.objects.with_feature('event_rules')
"""
if feature not in registry['model_features']:
raise KeyError(
f"{feature} is not a registered model feature! Valid features are: {registry['model_features'].keys()}"
)
q = Q()
for app_label, models in registry['model_features'][feature].items():
q |= Q(app_label=app_label, model__in=models)
return self.get_queryset().filter(q)
return self.get_queryset().filter(features__contains=[feature])
class ObjectType(ContentType):
"""
Wrap Django's native ContentType model to use our custom manager.
"""
contenttype_ptr = models.OneToOneField(
on_delete=models.CASCADE,
to='contenttypes.ContentType',
parent_link=True,
primary_key=True,
serialize=False,
related_name='object_type',
)
public = models.BooleanField(
default=False,
)
features = ArrayField(
base_field=models.CharField(max_length=50),
default=list,
)
objects = ObjectTypeManager()
class Meta:
proxy = True
verbose_name = _('object type')
verbose_name_plural = _('object types')
@property
def app_labeled_name(self):

View File

@ -20,6 +20,7 @@ from core.choices import JobStatusChoices
from core.dataclasses import JobLogEntry
from core.models import ObjectType
from core.signals import job_end, job_start
from netbox.models.features import has_feature
from utilities.json import JobLogDecoder
from utilities.querysets import RestrictedQuerySet
from utilities.rqworker import get_queue_for_model
@ -148,7 +149,7 @@ class Job(models.Model):
super().clean()
# Validate the assigned object type
if self.object_type and self.object_type not in ObjectType.objects.with_feature('jobs'):
if self.object_type and not has_feature(self.object_type, 'jobs'):
raise ValidationError(
_("Jobs cannot be assigned to this object type ({type}).").format(type=self.object_type)
)

View File

@ -1,20 +1,21 @@
import logging
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.core.exceptions import ObjectDoesNotExist, ValidationError
from django.db.models.fields.reverse_related import ManyToManyRel, ManyToOneRel
from django.db.models.signals import m2m_changed, post_save, pre_delete
from django.db.models.signals import m2m_changed, post_migrate, post_save, pre_delete
from django.dispatch import receiver, Signal
from django.utils.translation import gettext_lazy as _
from django_prometheus.models import model_deletes, model_inserts, model_updates
from core.choices import JobStatusChoices, ObjectChangeActionChoices
from core.events import *
from core.models import ObjectType
from extras.events import enqueue_event
from extras.utils import run_validators
from netbox.config import get_config
from netbox.context import current_request, events_queue
from netbox.models.features import ChangeLoggingMixin
from netbox.models.features import ChangeLoggingMixin, get_model_features, model_is_public
from utilities.exceptions import AbortRequest
from .models import ConfigRevision, DataSource, ObjectChange
@ -38,6 +39,37 @@ post_sync = Signal()
clear_events = Signal()
#
# Object types
#
@receiver(post_migrate)
def update_object_types(sender, **kwargs):
"""
Create or update the corresponding ObjectType for each model within the migrated app.
"""
for model in sender.get_models():
app_label, model_name = model._meta.label_lower.split('.')
# Determine whether model is public and its supported features
is_public = model_is_public(model)
features = get_model_features(model)
# Create/update the ObjectType for the model
try:
ot = ObjectType.objects.get_by_natural_key(app_label=app_label, model=model_name)
ot.public = is_public
ot.features = features
ot.save()
except ObjectDoesNotExist:
ObjectType.objects.create(
app_label=app_label,
model=model_name,
public=is_public,
features=features,
)
#
# Change logging & event handling
#

View File

@ -1,6 +1,7 @@
import itertools
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.db import models
from django.dispatch import Signal
@ -479,13 +480,13 @@ class CablePath(models.Model):
def origin_type(self):
if self.path:
ct_id, _ = decompile_path_node(self.path[0][0])
return ObjectType.objects.get_for_id(ct_id)
return ContentType.objects.get_for_id(ct_id)
@property
def destination_type(self):
if self.is_complete:
ct_id, _ = decompile_path_node(self.path[-1][0])
return ObjectType.objects.get_for_id(ct_id)
return ContentType.objects.get_for_id(ct_id)
@property
def _path_decompiled(self):

View File

@ -4,6 +4,7 @@ import yaml
from functools import cached_property
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.core.files.storage import default_storage
from django.core.validators import MaxValueValidator, MinValueValidator
@ -15,7 +16,6 @@ from django.urls import reverse
from django.utils.safestring import mark_safe
from django.utils.translation import gettext_lazy as _
from core.models import ObjectType
from dcim.choices import *
from dcim.constants import *
from dcim.fields import MACAddressField
@ -1328,7 +1328,7 @@ class MACAddress(PrimaryModel):
super().clean()
if self._original_assigned_object_id and self._original_assigned_object_type_id:
assigned_object = self.assigned_object
ct = ObjectType.objects.get_for_id(self._original_assigned_object_type_id)
ct = ContentType.objects.get_for_id(self._original_assigned_object_type_id)
original_assigned_object = ct.get_object_for_this_type(pk=self._original_assigned_object_id)
if (

View File

@ -11,7 +11,7 @@ from django_rq import get_queue
from core.events import *
from netbox.config import get_config
from netbox.constants import RQ_QUEUE_DEFAULT
from netbox.registry import registry
from netbox.models.features import has_feature
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 not has_feature(instance, 'event_rules'):
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}'

View File

@ -0,0 +1,56 @@
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0017_concrete_objecttype'),
('extras', '0130_imageattachment_description'),
]
operations = [
migrations.AlterField(
model_name='Tag',
name='object_types',
field=models.ManyToManyField(blank=True, related_name='+', to='core.objecttype'),
),
migrations.AlterField(
model_name='CustomField',
name='related_object_type',
field=models.ForeignKey(
blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, to='core.objecttype'
),
),
migrations.AlterField(
model_name='CustomField',
name='object_types',
field=models.ManyToManyField(related_name='custom_fields', to='core.objecttype'),
),
migrations.AlterField(
model_name='EventRule',
name='object_types',
field=models.ManyToManyField(related_name='event_rules', to='core.objecttype'),
),
migrations.AlterField(
model_name='CustomLink',
name='object_types',
field=models.ManyToManyField(related_name='custom_links', to='core.objecttype'),
),
migrations.AlterField(
model_name='ExportTemplate',
name='object_types',
field=models.ManyToManyField(related_name='export_templates', to='core.objecttype'),
),
migrations.AlterField(
model_name='SavedFilter',
name='object_types',
field=models.ManyToManyField(related_name='saved_filters', to='core.objecttype'),
),
migrations.AlterField(
model_name='TableConfig',
name='object_type',
field=models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, related_name='table_configs', to='core.objecttype'
),
),
]

View File

@ -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:

View File

@ -12,17 +12,16 @@ from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from rest_framework.utils.encoders import JSONEncoder
from core.models import ObjectType
from extras.choices import *
from extras.conditions import ConditionSet, InvalidCondition
from extras.constants import *
from extras.utils import image_upload
from extras.models.mixins import RenderTemplateMixin
from extras.utils import image_upload
from netbox.config import get_config
from netbox.events import get_event_type_choices
from netbox.models import ChangeLoggedModel
from netbox.models.features import (
CloningMixin, CustomFieldsMixin, CustomLinksMixin, ExportTemplatesMixin, SyncedDataMixin, TagsMixin
CloningMixin, CustomFieldsMixin, CustomLinksMixin, ExportTemplatesMixin, SyncedDataMixin, TagsMixin, has_feature
)
from utilities.html import clean_html
from utilities.jinja2 import render_jinja2
@ -707,7 +706,7 @@ class ImageAttachment(ChangeLoggedModel):
super().clean()
# Validate the assigned object type
if self.object_type not in ObjectType.objects.with_feature('image_attachments'):
if not has_feature(self.object_type, 'image_attachments'):
raise ValidationError(
_("Image attachments cannot be assigned to this object type ({type}).").format(type=self.object_type)
)
@ -807,7 +806,7 @@ class JournalEntry(CustomFieldsMixin, CustomLinksMixin, TagsMixin, ExportTemplat
super().clean()
# Validate the assigned object type
if self.assigned_object_type not in ObjectType.objects.with_feature('journaling'):
if not has_feature(self.assigned_object_type, 'journaling'):
raise ValidationError(
_("Journaling is not supported for this object type ({type}).").format(type=self.assigned_object_type)
)
@ -863,7 +862,7 @@ class Bookmark(models.Model):
super().clean()
# Validate the assigned object type
if self.object_type not in ObjectType.objects.with_feature('bookmarks'):
if not has_feature(self.object_type, 'bookmarks'):
raise ValidationError(
_("Bookmarks cannot be assigned to this object type ({type}).").format(type=self.object_type)
)

View File

@ -7,9 +7,9 @@ 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.querysets import NotificationQuerySet
from netbox.models import ChangeLoggedModel
from netbox.models.features import has_feature
from netbox.registry import registry
from users.models import User
from utilities.querysets import RestrictedQuerySet
@ -94,7 +94,7 @@ class Notification(models.Model):
super().clean()
# Validate the assigned object type
if self.object_type not in ObjectType.objects.with_feature('notifications'):
if not has_feature(self.object_type, 'notifications'):
raise ValidationError(
_("Objects of this type ({type}) do not support notifications.").format(type=self.object_type)
)
@ -235,7 +235,7 @@ class Subscription(models.Model):
super().clean()
# Validate the assigned object type
if self.object_type not in ObjectType.objects.with_feature('notifications'):
if not has_feature(self.object_type, 'notifications'):
raise ValidationError(
_("Objects of this type ({type}) do not support notifications.").format(type=self.object_type)
)

View File

@ -8,7 +8,7 @@ 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.models.features import has_feature
from netbox.signals import post_clean
from utilities.exceptions import AbortRequest
from .models import CustomField, TaggedItem
@ -150,17 +150,25 @@ 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, []):
if not has_feature(instance, 'notifications'):
return
ct = ContentType.objects.get_for_model(instance)
# 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=ct,
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=ct,
object_id=instance.pk,
user__in=subscribed_users
).delete()
# Create Notifications for Subscribers
Notification.objects.bulk_create([

View File

@ -1,5 +1,6 @@
import netaddr
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.db import models
from django.db.models import F
@ -7,7 +8,6 @@ from django.db.models.functions import Cast
from django.utils.functional import cached_property
from django.utils.translation import gettext_lazy as _
from core.models import ObjectType
from dcim.models.mixins import CachedScopeMixin
from ipam.choices import *
from ipam.constants import *
@ -917,7 +917,7 @@ class IPAddress(ContactsMixin, PrimaryModel):
if self._original_assigned_object_id and self._original_assigned_object_type_id:
parent = getattr(self.assigned_object, 'parent_object', None)
ct = ObjectType.objects.get_for_id(self._original_assigned_object_type_id)
ct = ContentType.objects.get_for_id(self._original_assigned_object_type_id)
original_assigned_object = ct.get_object_for_this_type(pk=self._original_assigned_object_id)
original_parent = getattr(original_assigned_object, 'parent_object', None)

View File

@ -3,6 +3,7 @@ from collections import defaultdict
from functools import cached_property
from django.contrib.contenttypes.fields import GenericRelation
from django.contrib.contenttypes.models import ContentType
from django.core.validators import ValidationError
from django.db import models
from django.db.models import Q
@ -32,12 +33,16 @@ __all__ = (
'CustomValidationMixin',
'EventRulesMixin',
'ExportTemplatesMixin',
'FEATURES_MAP',
'ImageAttachmentsMixin',
'JobsMixin',
'JournalingMixin',
'NotificationsMixin',
'SyncedDataMixin',
'TagsMixin',
'get_model_features',
'has_feature',
'model_is_public',
'register_models',
)
@ -633,11 +638,38 @@ FEATURES_MAP = {
'tags': TagsMixin,
}
# TODO: Remove in NetBox v4.5
registry['model_features'].update({
feature: defaultdict(set) for feature in FEATURES_MAP.keys()
})
def model_is_public(model):
return not getattr(model, '_netbox_private', False)
def get_model_features(model):
return [
feature for feature, cls in FEATURES_MAP.items() if issubclass(model, cls)
]
def has_feature(model_or_ct, feature):
"""
Returns True if the model supports the specified feature.
"""
# If an ObjectType was passed, we can use it directly
if type(model_or_ct) is ObjectType:
ot = model_or_ct
# If a ContentType was passed, resolve its model class
elif type(model_or_ct) is ContentType:
ot = ObjectType.objects.get_for_model(model_or_ct.model_class())
# For anything else, look up the ObjectType
else:
ot = ObjectType.objects.get_for_model(model_or_ct)
return feature in ot.features
def register_models(*models):
"""
Register one or more models in NetBox. This entails:
@ -653,10 +685,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)

View File

@ -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.assertTrue(
ObjectType.objects.filter(app_label='dummy_plugin', model='dummymodel').exists()
)
def test_models(self):
from netbox.tests.dummy_plugin.models import DummyModel

View File

@ -4,9 +4,8 @@ from django.db import models
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from core.models import ObjectType
from netbox.models import ChangeLoggedModel, NestedGroupModel, OrganizationalModel, PrimaryModel
from netbox.models.features import CustomFieldsMixin, ExportTemplatesMixin, TagsMixin
from netbox.models.features import CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, has_feature
from tenancy.choices import *
__all__ = (
@ -151,7 +150,7 @@ class ContactAssignment(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, Chan
super().clean()
# Validate the assigned object type
if self.object_type not in ObjectType.objects.with_feature('contacts'):
if not has_feature(self.object_type, 'contacts'):
raise ValidationError(
_("Contacts cannot be assigned to this object type ({type}).").format(type=self.object_type)
)

View File

@ -0,0 +1,17 @@
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0017_concrete_objecttype'),
('users', '0009_update_group_perms'),
]
operations = [
migrations.AlterField(
model_name='ObjectPermission',
name='object_types',
field=models.ManyToManyField(related_name='object_permissions', to='core.objecttype'),
),
]