ObjectTypeManager should not inherit from ContentTypeManager

This commit is contained in:
Jeremy Stretch 2025-07-24 08:14:09 -04:00
parent 0d8b6c78ae
commit 34d9ecb7f3
4 changed files with 37 additions and 36 deletions

View File

@ -2,8 +2,6 @@ import django.contrib.postgres.fields
import django.db.models.deletion import django.db.models.deletion
from django.db import migrations, models from django.db import migrations, models
import core.models.contenttypes
def populate_object_types(apps, schema_editor): def populate_object_types(apps, schema_editor):
""" """
@ -69,9 +67,7 @@ class Migration(migrations.Migration):
'verbose_name_plural': 'object types', 'verbose_name_plural': 'object types',
}, },
bases=('contenttypes.contenttype',), bases=('contenttypes.contenttype',),
managers=[ managers=[],
('objects', core.models.contenttypes.ObjectTypeManager()),
],
), ),
# Create an ObjectType record for each ContentType # Create an ObjectType record for each ContentType
migrations.RunPython( migrations.RunPython(

View File

@ -1,4 +1,4 @@
from django.contrib.contenttypes.models import ContentType, ContentTypeManager from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.fields import ArrayField from django.contrib.postgres.fields import ArrayField
from django.core.exceptions import ObjectDoesNotExist from django.core.exceptions import ObjectDoesNotExist
from django.db import models from django.db import models
@ -29,7 +29,7 @@ class ObjectTypeQuerySet(models.QuerySet):
return super().create(**kwargs) return super().create(**kwargs)
class ObjectTypeManager(ContentTypeManager): class ObjectTypeManager(models.Manager):
def get_queryset(self): def get_queryset(self):
return ObjectTypeQuerySet(self.model, using=self._db) return ObjectTypeQuerySet(self.model, using=self._db)
@ -37,33 +37,35 @@ class ObjectTypeManager(ContentTypeManager):
def create(self, **kwargs): def create(self, **kwargs):
return self.get_queryset().create(**kwargs) return self.get_queryset().create(**kwargs)
def get_for_model(self, model, for_concrete_model=True): def _get_opts(self, model, for_concrete_model):
""" if for_concrete_model:
Return the ContentType object for a given model, creating the model = model._meta.concrete_model
ContentType if necessary. Lookups are cached so that subsequent lookups return model._meta
for the same model don't hit the database.
""" def get_by_natural_key(self, app_label, model):
opts = self._get_opts(model, for_concrete_model) return self.get(app_label=app_label, model=model)
try:
return self._get_from_cache(opts) def get_for_id(self, id):
except KeyError: return self.get(pk=id)
pass
def get_for_model(self, model, for_concrete_model=True):
from netbox.models.features import get_model_features, model_is_public
opts = self._get_opts(model, for_concrete_model)
# The ContentType entry was not found in the cache, therefore we
# proceed to load or create it.
try: try:
# Start with get() and not get_or_create() in order to use # Start with get() and not get_or_create() in order to use
# the db_for_read (see #20401). # the db_for_read (see #20401).
ct = self.get(app_label=opts.app_label, model=opts.model_name) ot = self.get(app_label=opts.app_label, model=opts.model_name)
except self.model.DoesNotExist: except self.model.DoesNotExist:
# Not found in the database; we proceed to create it. This time # Not found in the database; we proceed to create it. This time
# use get_or_create to take care of any race conditions. # use get_or_create to take care of any race conditions.
ct, __ = self.get_or_create( ot, __ = self.get_or_create(
app_label=opts.app_label, app_label=opts.app_label,
model=opts.model_name, model=opts.model_name,
public=model_is_public(model),
features=get_model_features(model.__class__),
) )
self._add_to_cache(self.db, ct) return ot
return ct
def public(self): def public(self):
""" """

View File

@ -16,7 +16,7 @@ from extras.events import enqueue_event
from extras.utils import run_validators from extras.utils import run_validators
from netbox.config import get_config from netbox.config import get_config
from netbox.context import current_request, events_queue from netbox.context import current_request, events_queue
from netbox.models.features import ChangeLoggingMixin, FEATURES_MAP from netbox.models.features import ChangeLoggingMixin, get_model_features, model_is_public
from utilities.exceptions import AbortRequest from utilities.exceptions import AbortRequest
from .models import ConfigRevision, DataSource, ObjectChange from .models import ConfigRevision, DataSource, ObjectChange
@ -40,16 +40,6 @@ post_sync = Signal()
clear_events = Signal() clear_events = Signal()
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)
]
# #
# Object types # Object types
# #

View File

@ -40,7 +40,9 @@ __all__ = (
'NotificationsMixin', 'NotificationsMixin',
'SyncedDataMixin', 'SyncedDataMixin',
'TagsMixin', 'TagsMixin',
'get_model_features',
'has_feature', 'has_feature',
'model_is_public',
'register_models', 'register_models',
) )
@ -642,13 +644,24 @@ registry['model_features'].update({
}) })
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, feature): def has_feature(model, feature):
""" """
Returns True if the model supports the specified feature. Returns True if the model supports the specified feature.
""" """
if type(model) is ContentType: if type(model) is ContentType:
model = model.model_class() model = model.model_class()
return feature in ObjectType.objects.get_for_model(model).features ot = ObjectType.objects.get_for_model(model)
return feature in ot.features
def register_models(*models): def register_models(*models):