diff --git a/netbox/core/migrations/0008_contenttype_proxy.py b/netbox/core/migrations/0008_contenttype_proxy.py index 9acaf3ad7..a0af81813 100644 --- a/netbox/core/migrations/0008_contenttype_proxy.py +++ b/netbox/core/migrations/0008_contenttype_proxy.py @@ -1,4 +1,4 @@ -import core.models.contenttypes +import core.models.object_types from django.db import migrations @@ -19,7 +19,7 @@ class Migration(migrations.Migration): }, bases=('contenttypes.contenttype',), managers=[ - ('objects', core.models.contenttypes.ObjectTypeManager()), + ('objects', core.models.object_types.ObjectTypeManager()), ], ), ] diff --git a/netbox/core/migrations/0018_concrete_objecttype.py b/netbox/core/migrations/0018_concrete_objecttype.py new file mode 100644 index 000000000..4e227fe7a --- /dev/null +++ b/netbox/core/migrations/0018_concrete_objecttype.py @@ -0,0 +1,63 @@ +import django.contrib.postgres.fields +import django.contrib.postgres.indexes +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('contenttypes', '0002_remove_content_type_name'), + ('core', '0017_objectchange_message'), + ] + + 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', + 'ordering': ('app_label', 'model'), + 'indexes': [ + django.contrib.postgres.indexes.GinIndex( + fields=['features'], + name='core_object_feature_aec4de_gin' + ), + ] + }, + bases=('contenttypes.contenttype',), + managers=[], + ), + ] diff --git a/netbox/core/models/__init__.py b/netbox/core/models/__init__.py index db00e67aa..27eb3a92b 100644 --- a/netbox/core/models/__init__.py +++ b/netbox/core/models/__init__.py @@ -1,4 +1,4 @@ -from .contenttypes import * +from .object_types import * from .change_logging import * from .config import * from .data import * diff --git a/netbox/core/models/change_logging.py b/netbox/core/models/change_logging.py index 819b1b2b3..a011c457f 100644 --- a/netbox/core/models/change_logging.py +++ b/netbox/core/models/change_logging.py @@ -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', @@ -124,7 +124,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 diff --git a/netbox/core/models/contenttypes.py b/netbox/core/models/contenttypes.py index a7d5c91af..3d5c5e8b2 100644 --- a/netbox/core/models/contenttypes.py +++ b/netbox/core/models/contenttypes.py @@ -1,78 +1,3 @@ -from django.contrib.contenttypes.models import ContentType, ContentTypeManager -from django.db.models import Q - -from netbox.plugins import PluginConfig -from netbox.registry import registry -from utilities.string import title - -__all__ = ( - 'ObjectType', - 'ObjectTypeManager', -) - - -class ObjectTypeManager(ContentTypeManager): - - 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. - """ - 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) - - 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 - - ContentType.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) - - -class ObjectType(ContentType): - """ - Wrap Django's native ContentType model to use our custom manager. - """ - objects = ObjectTypeManager() - - class Meta: - proxy = True - - @property - def app_labeled_name(self): - # Override ContentType's "app | model" representation style. - return f"{self.app_verbose_name} > {title(self.model_verbose_name)}" - - @property - def app_verbose_name(self): - if model := self.model_class(): - return model._meta.app_config.verbose_name - - @property - def model_verbose_name(self): - if model := self.model_class(): - return model._meta.verbose_name - - @property - def model_verbose_name_plural(self): - if model := self.model_class(): - return model._meta.verbose_name_plural - - @property - def is_plugin_model(self): - if not (model := self.model_class()): - return # Return null if model class is invalid - return isinstance(model._meta.app_config, PluginConfig) +# TODO: Remove this module in NetBox v4.5 +# Provided for backward compatibility +from .object_types import * diff --git a/netbox/core/models/jobs.py b/netbox/core/models/jobs.py index 050f71921..0377ffbb1 100644 --- a/netbox/core/models/jobs.py +++ b/netbox/core/models/jobs.py @@ -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) ) diff --git a/netbox/core/models/object_types.py b/netbox/core/models/object_types.py new file mode 100644 index 000000000..e8a4f7202 --- /dev/null +++ b/netbox/core/models/object_types.py @@ -0,0 +1,205 @@ +from collections import defaultdict + +from django.contrib.contenttypes.models import ContentType +from django.contrib.postgres.fields import ArrayField +from django.contrib.postgres.indexes import GinIndex +from django.core.exceptions import ObjectDoesNotExist +from django.db import models +from django.db.models import Q +from django.utils.translation import gettext as _ + +from netbox.plugins import PluginConfig +from netbox.registry import registry +from utilities.string import title + +__all__ = ( + 'ObjectType', + 'ObjectTypeManager', + 'ObjectTypeQuerySet', +) + + +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 get_for_models(self, *models, for_concrete_models=True): + """ + Retrieve or create the ObjectTypes for multiple models, returning a mapping {model: ObjectType}. + + This method exists to provide parity with ContentTypeManager. + """ + from netbox.models.features import get_model_features, model_is_public + results = {} + + # Compile the model and options mappings + needed_models = defaultdict(set) + needed_opts = defaultdict(list) + for model in models: + opts = self._get_opts(model, for_concrete_models) + needed_models[opts.app_label].add(opts.model_name) + needed_opts[(opts.app_label, opts.model_name)].append(model) + + # Fetch existing ObjectType from the database + condition = Q( + *( + Q(('app_label', app_label), ('model__in', model_names)) + for app_label, model_names in needed_models.items() + ), + _connector=Q.OR, + ) + for ot in self.filter(condition): + opts_models = needed_opts.pop((ot.app_label, ot.model), []) + for model in opts_models: + results[model] = ot + + # Create any missing ObjectTypes + for (app_label, model_name), opts_models in needed_opts.items(): + for model in opts_models: + results[model] = self.create( + app_label=app_label, + model=model_name, + public=model_is_public(model), + features=get_model_features(model.__class__), + ) + + return results + + def public(self): + """ + 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. + """ + return self.get_queryset().filter(public=True) + + def with_feature(self, feature): + """ + Return ObjectTypes only for models which support the given feature. + + 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()}" + ) + 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: + verbose_name = _('object type') + verbose_name_plural = _('object types') + ordering = ('app_label', 'model') + indexes = [ + GinIndex(fields=['features']), + ] + + @property + def app_labeled_name(self): + # Override ContentType's "app | model" representation style. + return f"{self.app_verbose_name} > {title(self.model_verbose_name)}" + + @property + def app_verbose_name(self): + if model := self.model_class(): + return model._meta.app_config.verbose_name + + @property + def model_verbose_name(self): + if model := self.model_class(): + return model._meta.verbose_name + + @property + def model_verbose_name_plural(self): + if model := self.model_class(): + return model._meta.verbose_name_plural + + @property + def is_plugin_model(self): + if not (model := self.model_class()): + return # Return null if model class is invalid + return isinstance(model._meta.app_config, PluginConfig) diff --git a/netbox/core/signals.py b/netbox/core/signals.py index 3d0317011..0b8490dcb 100644 --- a/netbox/core/signals.py +++ b/netbox/core/signals.py @@ -2,9 +2,9 @@ import logging from threading import local 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.core.signals import request_finished from django.utils.translation import gettext_lazy as _ @@ -12,11 +12,12 @@ 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 @@ -40,6 +41,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 # diff --git a/netbox/core/tests/test_models.py b/netbox/core/tests/test_models.py index ff71c2e88..28225c7a6 100644 --- a/netbox/core/tests/test_models.py +++ b/netbox/core/tests/test_models.py @@ -1,7 +1,10 @@ +from django.contrib.contenttypes.models import ContentType +from django.core.exceptions import ObjectDoesNotExist from django.test import TestCase -from core.models import DataSource +from core.models import DataSource, ObjectType from core.choices import ObjectChangeActionChoices +from dcim.models import Site, Location, Device from netbox.constants import CENSOR_TOKEN, CENSOR_TOKEN_CHANGED @@ -120,3 +123,80 @@ class DataSourceChangeLoggingTestCase(TestCase): self.assertEqual(objectchange.prechange_data['parameters']['password'], CENSOR_TOKEN) self.assertEqual(objectchange.postchange_data['parameters']['username'], 'username2') self.assertEqual(objectchange.postchange_data['parameters']['password'], CENSOR_TOKEN) + + +class ObjectTypeTest(TestCase): + + def test_create(self): + """ + Test that an ObjectType created for a given app_label & model name will be automatically assigned to + the appropriate ContentType. + """ + kwargs = { + 'app_label': 'foo', + 'model': 'bar', + } + ct = ContentType.objects.create(**kwargs) + ot = ObjectType.objects.create(**kwargs) + self.assertEqual(ot.contenttype_ptr, ct) + + def test_get_by_natural_key(self): + """ + Test that get_by_natural_key() returns the appropriate ObjectType. + """ + self.assertEqual( + ObjectType.objects.get_by_natural_key('dcim', 'site'), + ObjectType.objects.get(app_label='dcim', model='site') + ) + with self.assertRaises(ObjectDoesNotExist): + ObjectType.objects.get_by_natural_key('foo', 'bar') + + def test_get_for_id(self): + """ + Test that get_by_id() returns the appropriate ObjectType. + """ + ot = ObjectType.objects.get_by_natural_key('dcim', 'site') + self.assertEqual( + ObjectType.objects.get_for_id(ot.pk), + ObjectType.objects.get(pk=ot.pk) + ) + with self.assertRaises(ObjectDoesNotExist): + ObjectType.objects.get_for_id(0) + + def test_get_for_model(self): + """ + Test that get_by_model() returns the appropriate ObjectType. + """ + self.assertEqual( + ObjectType.objects.get_for_model(Site), + ObjectType.objects.get_by_natural_key('dcim', 'site') + ) + + def test_get_for_models(self): + """ + Test that get_by_models() returns the appropriate ObjectType mapping. + """ + self.assertEqual( + ObjectType.objects.get_for_models(Site, Location, Device), + { + Site: ObjectType.objects.get_by_natural_key('dcim', 'site'), + Location: ObjectType.objects.get_by_natural_key('dcim', 'location'), + Device: ObjectType.objects.get_by_natural_key('dcim', 'device'), + } + ) + + def test_public(self): + """ + Test that public() returns only ObjectTypes for public models. + """ + public_ots = ObjectType.objects.public() + self.assertIn(ObjectType.objects.get_by_natural_key('dcim', 'site'), public_ots) + self.assertNotIn(ObjectType.objects.get_by_natural_key('extras', 'taggeditem'), public_ots) + + def test_with_feature(self): + """ + Test that with_feature() returns only ObjectTypes for models which support the specified feature. + """ + bookmarks_ots = ObjectType.objects.with_feature('bookmarks') + self.assertIn(ObjectType.objects.get_by_natural_key('dcim', 'site'), bookmarks_ots) + self.assertNotIn(ObjectType.objects.get_by_natural_key('dcim', 'cabletermination'), bookmarks_ots) diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index 0a28d5acb..32e26f982 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -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): diff --git a/netbox/dcim/models/devices.py b/netbox/dcim/models/devices.py index 264c1e025..78fd881a7 100644 --- a/netbox/dcim/models/devices.py +++ b/netbox/dcim/models/devices.py @@ -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 ( diff --git a/netbox/extras/events.py b/netbox/extras/events.py index d7c642c4e..f8447fdb2 100644 --- a/netbox/extras/events.py +++ b/netbox/extras/events.py @@ -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}' diff --git a/netbox/extras/migrations/0111_rename_content_types.py b/netbox/extras/migrations/0111_rename_content_types.py index a9f80b146..8a5db53ff 100644 --- a/netbox/extras/migrations/0111_rename_content_types.py +++ b/netbox/extras/migrations/0111_rename_content_types.py @@ -24,7 +24,7 @@ class Migration(migrations.Migration): model_name='customfield', name='object_type', field=models.ForeignKey( - blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, to='core.objecttype' + blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, to='contenttypes.contenttype' ), ), migrations.RunSQL(( diff --git a/netbox/extras/migrations/0128_tableconfig.py b/netbox/extras/migrations/0128_tableconfig.py index e6d45199d..98048ee27 100644 --- a/netbox/extras/migrations/0128_tableconfig.py +++ b/netbox/extras/migrations/0128_tableconfig.py @@ -37,7 +37,9 @@ class Migration(migrations.Migration): ( 'object_type', models.ForeignKey( - on_delete=django.db.models.deletion.CASCADE, related_name='table_configs', to='core.objecttype' + on_delete=django.db.models.deletion.CASCADE, + related_name='table_configs', + to='contenttypes.contenttype' ), ), ( diff --git a/netbox/extras/migrations/0131_concrete_objecttype.py b/netbox/extras/migrations/0131_concrete_objecttype.py new file mode 100644 index 000000000..6aed4d97d --- /dev/null +++ b/netbox/extras/migrations/0131_concrete_objecttype.py @@ -0,0 +1,42 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('contenttypes', '0002_remove_content_type_name'), + ('extras', '0130_imageattachment_description'), + ] + + operations = [ + migrations.AlterField( + model_name='customfield', + name='object_types', + field=models.ManyToManyField(related_name='custom_fields', to='contenttypes.contenttype'), + ), + migrations.AlterField( + model_name='customlink', + name='object_types', + field=models.ManyToManyField(related_name='custom_links', to='contenttypes.contenttype'), + ), + migrations.AlterField( + model_name='eventrule', + name='object_types', + field=models.ManyToManyField(related_name='event_rules', to='contenttypes.contenttype'), + ), + migrations.AlterField( + model_name='exporttemplate', + name='object_types', + field=models.ManyToManyField(related_name='export_templates', to='contenttypes.contenttype'), + ), + migrations.AlterField( + model_name='savedfilter', + name='object_types', + field=models.ManyToManyField(related_name='saved_filters', to='contenttypes.contenttype'), + ), + migrations.AlterField( + model_name='tag', + name='object_types', + field=models.ManyToManyField(blank=True, related_name='+', to='contenttypes.contenttype'), + ), + ] diff --git a/netbox/extras/models/configs.py b/netbox/extras/models/configs.py index 8d6b8d999..f92c66632 100644 --- a/netbox/extras/models/configs.py +++ b/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: diff --git a/netbox/extras/models/customfields.py b/netbox/extras/models/customfields.py index aeeb15728..88e6abc27 100644 --- a/netbox/extras/models/customfields.py +++ b/netbox/extras/models/customfields.py @@ -72,7 +72,7 @@ class CustomFieldManager(models.Manager.from_queryset(RestrictedQuerySet)): class CustomField(CloningMixin, ExportTemplatesMixin, ChangeLoggedModel): object_types = models.ManyToManyField( - to='core.ObjectType', + to='contenttypes.ContentType', related_name='custom_fields', help_text=_('The object(s) to which this field applies.') ) @@ -84,7 +84,7 @@ class CustomField(CloningMixin, ExportTemplatesMixin, ChangeLoggedModel): help_text=_('The type of data this custom field holds') ) related_object_type = models.ForeignKey( - to='core.ObjectType', + to='contenttypes.ContentType', on_delete=models.PROTECT, blank=True, null=True, diff --git a/netbox/extras/models/models.py b/netbox/extras/models/models.py index 2fdc1ffe3..fc69e692c 100644 --- a/netbox/extras/models/models.py +++ b/netbox/extras/models/models.py @@ -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 @@ -50,7 +49,7 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, ChangeLogged webhook or executing a custom script. """ object_types = models.ManyToManyField( - to='core.ObjectType', + to='contenttypes.ContentType', related_name='event_rules', verbose_name=_('object types'), help_text=_("The object(s) to which this rule applies.") @@ -299,7 +298,7 @@ class CustomLink(CloningMixin, ExportTemplatesMixin, ChangeLoggedModel): code to be rendered with an object as context. """ object_types = models.ManyToManyField( - to='core.ObjectType', + to='contenttypes.ContentType', related_name='custom_links', help_text=_('The object type(s) to which this link applies.') ) @@ -395,7 +394,7 @@ class CustomLink(CloningMixin, ExportTemplatesMixin, ChangeLoggedModel): class ExportTemplate(SyncedDataMixin, CloningMixin, ExportTemplatesMixin, ChangeLoggedModel, RenderTemplateMixin): object_types = models.ManyToManyField( - to='core.ObjectType', + to='contenttypes.ContentType', related_name='export_templates', help_text=_('The object type(s) to which this template applies.') ) @@ -460,7 +459,7 @@ class SavedFilter(CloningMixin, ExportTemplatesMixin, ChangeLoggedModel): A set of predefined keyword parameters that can be reused to filter for specific objects. """ object_types = models.ManyToManyField( - to='core.ObjectType', + to='contenttypes.ContentType', related_name='saved_filters', help_text=_('The object type(s) to which this filter applies.') ) @@ -540,7 +539,7 @@ class TableConfig(CloningMixin, ChangeLoggedModel): A saved configuration of columns and ordering which applies to a specific table. """ object_type = models.ForeignKey( - to='core.ObjectType', + to='contenttypes.ContentType', on_delete=models.CASCADE, related_name='table_configs', help_text=_("The table's object type"), @@ -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) ) diff --git a/netbox/extras/models/notifications.py b/netbox/extras/models/notifications.py index 44874a4c8..c8e8c4fd8 100644 --- a/netbox/extras/models/notifications.py +++ b/netbox/extras/models/notifications.py @@ -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) ) diff --git a/netbox/extras/models/tags.py b/netbox/extras/models/tags.py index b40327265..c05f44faf 100644 --- a/netbox/extras/models/tags.py +++ b/netbox/extras/models/tags.py @@ -35,7 +35,7 @@ class Tag(CloningMixin, ExportTemplatesMixin, ChangeLoggedModel, TagBase): blank=True, ) object_types = models.ManyToManyField( - to='core.ObjectType', + to='contenttypes.ContentType', related_name='+', blank=True, help_text=_("The object type(s) to which this tag can be applied.") diff --git a/netbox/extras/signals.py b/netbox/extras/signals.py index 10c3f73c5..7105c38b4 100644 --- a/netbox/extras/signals.py +++ b/netbox/extras/signals.py @@ -3,12 +3,11 @@ from django.db.models.signals import m2m_changed, post_save, pre_delete from django.dispatch import receiver from core.events import * -from core.models import ObjectType 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 @@ -82,7 +81,7 @@ def validate_assigned_tags(sender, instance, action, model, pk_set, **kwargs): """ if action != 'pre_add': return - ct = ObjectType.objects.get_for_model(instance) + ct = ContentType.objects.get_for_model(instance) # Retrieve any applied Tags that are restricted to certain object types for tag in model.objects.filter(pk__in=pk_set, object_types__isnull=False).prefetch_related('object_types'): if ct not in tag.object_types.all(): @@ -150,17 +149,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([ diff --git a/netbox/ipam/models/ip.py b/netbox/ipam/models/ip.py index db116e9e4..73c3310dc 100644 --- a/netbox/ipam/models/ip.py +++ b/netbox/ipam/models/ip.py @@ -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) diff --git a/netbox/netbox/constants.py b/netbox/netbox/constants.py index ad96a643d..75e0f4da6 100644 --- a/netbox/netbox/constants.py +++ b/netbox/netbox/constants.py @@ -1,3 +1,18 @@ +CORE_APPS = ( + 'account', + 'circuits', + 'core', + 'dcim', + 'extras', + 'ipam', + 'tenancy', + 'users', + 'utilities', + 'virtualization', + 'vpn', + 'wireless', +) + # RQ queue names RQ_QUEUE_DEFAULT = 'default' RQ_QUEUE_HIGH = 'high' diff --git a/netbox/netbox/models/features.py b/netbox/netbox/models/features.py index 893131336..b71eecaaa 100644 --- a/netbox/netbox/models/features.py +++ b/netbox/netbox/models/features.py @@ -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 @@ -16,7 +17,9 @@ from extras.choices import * from extras.constants import CUSTOMFIELD_EMPTY_VALUES from extras.utils import is_taggable from netbox.config import get_config +from netbox.constants import CORE_APPS from netbox.models.deletion import DeleteMixin +from netbox.plugins import PluginConfig from netbox.registry import registry from netbox.signals import post_clean from utilities.json import CustomFieldJSONEncoder @@ -32,12 +35,16 @@ __all__ = ( 'CustomValidationMixin', 'EventRulesMixin', 'ExportTemplatesMixin', + 'FEATURES_MAP', 'ImageAttachmentsMixin', 'JobsMixin', 'JournalingMixin', 'NotificationsMixin', 'SyncedDataMixin', 'TagsMixin', + 'get_model_features', + 'has_feature', + 'model_is_public', 'register_models', ) @@ -639,11 +646,46 @@ 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 True if the model is considered "public use;" otherwise return False. + + All non-core and non-plugin models are excluded. + """ + opts = model._meta + if opts.app_label not in CORE_APPS and not isinstance(opts.app_config, PluginConfig): + return False + 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: @@ -659,10 +701,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) diff --git a/netbox/netbox/tests/test_model_features.py b/netbox/netbox/tests/test_model_features.py new file mode 100644 index 000000000..190c177ea --- /dev/null +++ b/netbox/netbox/tests/test_model_features.py @@ -0,0 +1,53 @@ +from django.test import TestCase + +from core.models import AutoSyncRecord, DataSource +from extras.models import CustomLink +from netbox.models.features import get_model_features, has_feature, model_is_public +from netbox.tests.dummy_plugin.models import DummyModel +from taggit.models import Tag + + +class ModelFeaturesTestCase(TestCase): + + def test_model_is_public(self): + """ + Test that the is_public() utility function returns True for public models only. + """ + # Public model + self.assertFalse(hasattr(DataSource, '_netbox_private')) + self.assertTrue(model_is_public(DataSource)) + + # Private model + self.assertTrue(getattr(AutoSyncRecord, '_netbox_private')) + self.assertFalse(model_is_public(AutoSyncRecord)) + + # Plugin model + self.assertFalse(hasattr(DummyModel, '_netbox_private')) + self.assertTrue(model_is_public(DummyModel)) + + # Non-core model + self.assertFalse(hasattr(Tag, '_netbox_private')) + self.assertFalse(model_is_public(Tag)) + + def test_has_feature(self): + """ + Test the functionality of the has_feature() utility function. + """ + # Sanity checking + self.assertTrue(hasattr(DataSource, 'bookmarks'), "Invalid test?") + self.assertFalse(hasattr(AutoSyncRecord, 'bookmarks'), "Invalid test?") + + self.assertTrue(has_feature(DataSource, 'bookmarks')) + self.assertFalse(has_feature(AutoSyncRecord, 'bookmarks')) + + def test_get_model_features(self): + """ + Check that get_model_features() returns the expected features for a model. + """ + # Sanity checking + self.assertTrue(hasattr(CustomLink, 'clone'), "Invalid test?") + self.assertFalse(hasattr(CustomLink, 'bookmarks'), "Invalid test?") + + features = get_model_features(CustomLink) + self.assertIn('cloning', features) + self.assertNotIn('bookmarks', features) diff --git a/netbox/netbox/tests/test_plugins.py b/netbox/netbox/tests/test_plugins.py index 264c8e6f9..9f2033936 100644 --- a/netbox/netbox/tests/test_plugins.py +++ b/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.assertTrue( + ObjectType.objects.filter(app_label='dummy_plugin', model='dummymodel').exists() + ) def test_models(self): from netbox.tests.dummy_plugin.models import DummyModel diff --git a/netbox/tenancy/models/contacts.py b/netbox/tenancy/models/contacts.py index 34e444ee7..19ffb2b0b 100644 --- a/netbox/tenancy/models/contacts.py +++ b/netbox/tenancy/models/contacts.py @@ -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) ) diff --git a/netbox/users/migrations/0010_concrete_objecttype.py b/netbox/users/migrations/0010_concrete_objecttype.py new file mode 100644 index 000000000..436758846 --- /dev/null +++ b/netbox/users/migrations/0010_concrete_objecttype.py @@ -0,0 +1,17 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('contenttypes', '0002_remove_content_type_name'), + ('users', '0009_update_group_perms'), + ] + + operations = [ + migrations.AlterField( + model_name='objectpermission', + name='object_types', + field=models.ManyToManyField(related_name='object_permissions', to='contenttypes.contenttype'), + ), + ] diff --git a/netbox/users/models/permissions.py b/netbox/users/models/permissions.py index 772adcdb7..3ae8ff4c1 100644 --- a/netbox/users/models/permissions.py +++ b/netbox/users/models/permissions.py @@ -29,7 +29,7 @@ class ObjectPermission(models.Model): default=True ) object_types = models.ManyToManyField( - to='core.ObjectType', + to='contenttypes.ContentType', related_name='object_permissions' ) actions = ArrayField( diff --git a/netbox/utilities/testing/filtersets.py b/netbox/utilities/testing/filtersets.py index 260c092db..b8fb62821 100644 --- a/netbox/utilities/testing/filtersets.py +++ b/netbox/utilities/testing/filtersets.py @@ -1,19 +1,17 @@ -import django_filters from datetime import datetime, timezone from itertools import chain -from mptt.models import MPTTModel +import django_filters from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation from django.contrib.contenttypes.models import ContentType from django.db.models import ForeignKey, ManyToManyField, ManyToManyRel, ManyToOneRel, OneToOneRel from django.utils.module_loading import import_string +from mptt.models import MPTTModel from taggit.managers import TaggableManager from extras.filters import TagFilter from utilities.filters import ContentTypeFilter, TreeNodeMultipleChoiceFilter -from core.models import ObjectType - __all__ = ( 'BaseFilterSetTests', 'ChangeLoggedFilterSetTests', @@ -61,13 +59,6 @@ class BaseFilterSetTests: if field.related_model is ContentType: return [(None, None)] - # ForeignKeys to ObjectType need two filters: 'app.model' & PK - if field.related_model is ObjectType: - return [ - (filter_name, ContentTypeFilter), - (f'{filter_name}_id', django_filters.ModelMultipleChoiceFilter), - ] - # ForeignKey to an MPTT-enabled model if issubclass(field.related_model, MPTTModel) and field.model is not field.related_model: return [(f'{filter_name}_id', TreeNodeMultipleChoiceFilter)] @@ -79,8 +70,10 @@ class BaseFilterSetTests: filter_name = self.get_m2m_filter_name(field) filter_name = self.filter_name_map.get(filter_name, filter_name) - # ManyToManyFields to ObjectType need two filters: 'app.model' & PK - if field.related_model is ObjectType: + # ManyToManyFields to ContentType need two filters: 'app.model' & PK + if field.related_model is ContentType: + # Standardize on object_type for filter name even though it's technically a ContentType + filter_name = 'object_type' return [ (filter_name, ContentTypeFilter), (f'{filter_name}_id', django_filters.ModelMultipleChoiceFilter),