Coverage for core/models/object_types.py: 67%
115 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import inspect
2from collections import defaultdict
4from django.contrib.contenttypes.models import ContentType
5from django.contrib.postgres.fields import ArrayField
6from django.contrib.postgres.indexes import GinIndex
7from django.core.exceptions import ObjectDoesNotExist
8from django.db import connection, models
9from django.db.models import Q
10from django.utils.translation import gettext as _
12from netbox.context import query_cache
13from netbox.plugins import PluginConfig
14from netbox.registry import registry
15from utilities.string import title
17__all__ = (
18 'ObjectType',
19 'ObjectTypeManager',
20 'ObjectTypeQuerySet',
21)
24class ObjectTypeQuerySet(models.QuerySet):
26 def create(self, **kwargs):
27 # If attempting to create a new ObjectType for a given app_label & model, replace those kwargs
28 # with a reference to the ContentType (if one exists).
29 if (app_label := kwargs.get('app_label')) and (model := kwargs.get('model')): 29 ↛ 34line 29 didn't jump to line 34 because the condition on line 29 was always true
30 try:
31 kwargs['contenttype_ptr'] = ContentType.objects.get(app_label=app_label, model=model)
32 except ObjectDoesNotExist:
33 pass
34 return super().create(**kwargs)
37class ObjectTypeManager(models.Manager):
39 # TODO: Remove this in NetBox v5.0
40 # Cache the result of introspection to avoid repeated queries.
41 _table_exists = False
43 def get_queryset(self):
44 return ObjectTypeQuerySet(self.model, using=self._db)
46 def get_by_natural_key(self, app_label, model):
47 """
48 Retrieve an ObjectType by its application label & model name.
50 This method exists to provide parity with ContentTypeManager.
51 """
52 return self.get(app_label=app_label, model=model)
54 def get_for_id(self, id):
55 """
56 Retrieve an ObjectType by its primary key (numeric ID).
58 This method exists to provide parity with ContentTypeManager.
59 """
60 return self.get(pk=id)
62 def _get_opts(self, model, for_concrete_model):
63 if for_concrete_model: 63 ↛ 65line 63 didn't jump to line 65 because the condition on line 63 was always true
64 model = model._meta.concrete_model
65 return model._meta
67 def get_for_model(self, model, for_concrete_model=True):
68 """
69 Retrieve or create and return the ObjectType for a model.
70 """
71 from netbox.models.features import get_model_features, model_is_public
73 # Check the request cache before hitting the database
74 cache = query_cache.get()
75 if cache is not None:
76 if ot := cache['object_types'].get((model._meta.model, for_concrete_model)):
77 return ot
79 # TODO: Remove this in NetBox v5.0
80 # If the ObjectType table has not yet been provisioned (e.g. because we're in a pre-v4.4 migration),
81 # fall back to ContentType.
82 if not ObjectTypeManager._table_exists:
83 if 'core_objecttype' not in connection.introspection.table_names(): 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true
84 ct = ContentType.objects.get_for_model(model, for_concrete_model=for_concrete_model)
85 ct.features = get_model_features(ct.model_class())
86 return ct
87 ObjectTypeManager._table_exists = True
89 if not inspect.isclass(model):
90 model = model.__class__
91 opts = self._get_opts(model, for_concrete_model)
93 try:
94 # Use .get() instead of .get_or_create() initially to ensure db_for_read is honored (Django bug #20401).
95 ot = self.get(app_label=opts.app_label, model=opts.model_name)
96 except self.model.DoesNotExist:
97 # If the ObjectType doesn't exist, create it. (Use .get_or_create() to avoid race conditions.)
98 ot = self.get_or_create(
99 app_label=opts.app_label,
100 model=opts.model_name,
101 public=model_is_public(model),
102 features=get_model_features(model),
103 )[0]
105 # Populate the request cache to avoid redundant lookups
106 if cache is not None:
107 cache['object_types'][(model._meta.model, for_concrete_model)] = ot
109 return ot
111 def get_for_models(self, *models, for_concrete_models=True):
112 """
113 Retrieve or create the ObjectTypes for multiple models, returning a mapping {model: ObjectType}.
115 This method exists to provide parity with ContentTypeManager.
116 """
117 from netbox.models.features import get_model_features, model_is_public
118 results = {}
120 # Compile the model and options mappings
121 needed_models = defaultdict(set)
122 needed_opts = defaultdict(list)
123 for model in models:
124 if not inspect.isclass(model):
125 model = model.__class__
126 opts = self._get_opts(model, for_concrete_models)
127 needed_models[opts.app_label].add(opts.model_name)
128 needed_opts[(opts.app_label, opts.model_name)].append(model)
130 # Fetch existing ObjectType from the database
131 condition = Q(
132 *(
133 Q(('app_label', app_label), ('model__in', model_names))
134 for app_label, model_names in needed_models.items()
135 ),
136 _connector=Q.OR,
137 )
138 for ot in self.filter(condition):
139 opts_models = needed_opts.pop((ot.app_label, ot.model), [])
140 for model in opts_models:
141 results[model] = ot
143 # Create any missing ObjectTypes
144 for (app_label, model_name), opts_models in needed_opts.items():
145 for model in opts_models:
146 results[model] = self.create(
147 app_label=app_label,
148 model=model_name,
149 public=model_is_public(model),
150 features=get_model_features(model),
151 )
153 return results
155 def public(self):
156 """
157 Includes only ObjectTypes for "public" models.
159 Filter the base queryset to return only ObjectTypes corresponding to public models; those which are intended
160 for reference by other objects within the application.
161 """
162 return self.get_queryset().filter(public=True)
164 def with_feature(self, feature):
165 """
166 Return ObjectTypes only for models which support the given feature.
168 Only ObjectTypes which list the specified feature will be included. Supported features are declared in the
169 application registry under `registry["model_features"]`. For example, we can find all ObjectTypes for models
170 which support event rules with:
172 ObjectType.objects.with_feature('event_rules')
173 """
174 if feature not in registry['model_features']: 174 ↛ 175line 174 didn't jump to line 175 because the condition on line 174 was never true
175 raise KeyError(
176 f"{feature} is not a registered model feature! Valid features are: {registry['model_features'].keys()}"
177 )
178 return self.get_queryset().filter(features__contains=[feature])
181class ObjectType(ContentType):
182 """
183 Wrap Django's native ContentType model to use our custom manager.
184 """
185 contenttype_ptr = models.OneToOneField(
186 on_delete=models.CASCADE,
187 to='contenttypes.ContentType',
188 parent_link=True,
189 primary_key=True,
190 serialize=False,
191 related_name='object_type',
192 )
193 public = models.BooleanField(
194 default=False,
195 )
196 features = ArrayField(
197 base_field=models.CharField(max_length=50),
198 default=list,
199 )
201 objects = ObjectTypeManager()
203 class Meta:
204 verbose_name = _('object type')
205 verbose_name_plural = _('object types')
206 ordering = ('app_label', 'model')
207 indexes = [
208 GinIndex(fields=['features']),
209 ]
211 @property
212 def app_labeled_name(self):
213 # Override ContentType's "app | model" representation style.
214 return f"{self.app_verbose_name} > {title(self.model_verbose_name)}"
216 @property
217 def app_verbose_name(self):
218 if model := self.model_class(): 218 ↛ 220line 218 didn't jump to line 220 because the condition on line 218 was always true
219 return model._meta.app_config.verbose_name
220 return None
222 @property
223 def model_verbose_name(self):
224 if model := self.model_class(): 224 ↛ 226line 224 didn't jump to line 226 because the condition on line 224 was always true
225 return model._meta.verbose_name
226 return None
228 @property
229 def model_verbose_name_plural(self):
230 if model := self.model_class(): 230 ↛ 232line 230 didn't jump to line 232 because the condition on line 230 was always true
231 return model._meta.verbose_name_plural
232 return None
234 @property
235 def is_plugin_model(self):
236 if not (model := self.model_class()): 236 ↛ 237line 236 didn't jump to line 237 because the condition on line 236 was never true
237 return None # Return null if model class is invalid
238 return isinstance(model._meta.app_config, PluginConfig)