Coverage for netbox/models/features.py: 60%
328 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import json
2from collections import defaultdict
3from functools import cached_property
5from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
6from django.contrib.contenttypes.models import ContentType
7from django.core.validators import ValidationError
8from django.db import models, router, transaction
9from django.db.models import Q
10from django.utils import timezone
11from django.utils.translation import gettext_lazy as _
13from core.choices import JobStatusChoices, ObjectChangeActionChoices
14from core.models import ObjectType
15from extras.choices import *
16from extras.constants import CUSTOMFIELD_EMPTY_VALUES
17from extras.managers import NetBoxTaggableManager, NetBoxTaggableManagerField
18from extras.utils import is_taggable
19from netbox.config import get_config
20from netbox.constants import CORE_APPS, JOB_DELETE_BATCH_SIZE
21from netbox.models.deletion import DeleteMixin
22from netbox.plugins import PluginConfig
23from netbox.registry import registry
24from netbox.signals import post_clean
25from netbox.utils import register_model_feature
26from utilities.json import CustomFieldJSONEncoder
27from utilities.permissions import ModelAction, register_model_actions
28from utilities.serialization import serialize_object
30__all__ = (
31 'BookmarksMixin',
32 'ChangeLoggingMixin',
33 'CloningMixin',
34 'ContactsMixin',
35 'CustomFieldsMixin',
36 'CustomLinksMixin',
37 'CustomValidationMixin',
38 'EventRulesMixin',
39 'ExportTemplatesMixin',
40 'ImageAttachmentsMixin',
41 'JobsMixin',
42 'JournalingMixin',
43 'NotificationsMixin',
44 'SyncedDataMixin',
45 'TagsMixin',
46 'batch_delete_jobs',
47 'get_model_features',
48 'has_feature',
49 'model_is_public',
50 'register_models',
51)
54#
55# Feature mixins
56#
58class ChangeLoggingMixin(DeleteMixin, models.Model):
59 """
60 Provides change logging support for a model. Adds the `created` and `last_updated` fields.
61 """
62 created = models.DateTimeField(
63 verbose_name=_('created'),
64 auto_now_add=True,
65 blank=True,
66 null=True
67 )
68 last_updated = models.DateTimeField(
69 verbose_name=_('last updated'),
70 auto_now=True,
71 blank=True,
72 null=True
73 )
75 class Meta:
76 abstract = True
78 def __init__(self, *args, **kwargs):
79 changelog_message = kwargs.pop('changelog_message', None)
80 super().__init__(*args, **kwargs)
81 self._changelog_message = changelog_message
83 def serialize_object(self, exclude=None):
84 """
85 Return a JSON representation of the instance. Models can override this method to replace or extend the default
86 serialization logic provided by the `serialize_object()` utility function.
88 Args:
89 exclude: An iterable of attribute names to omit from the serialized output
90 """
91 return serialize_object(self, exclude=exclude or [])
93 def snapshot(self):
94 """
95 Save a snapshot of the object's current state in preparation for modification. The snapshot is saved as
96 `_prechange_snapshot` on the instance.
97 """
98 exclude_fields = []
99 if get_config().CHANGELOG_SKIP_EMPTY_CHANGES: 99 ↛ 102line 99 didn't jump to line 102 because the condition on line 99 was always true
100 exclude_fields = ['last_updated',]
102 self._prechange_snapshot = self.serialize_object(exclude=exclude_fields)
103 snapshot.alters_data = True
105 def to_objectchange(self, action):
106 """
107 Return a new ObjectChange representing a change made to this object. This will typically be called automatically
108 by ChangeLoggingMiddleware.
109 """
110 # TODO: Fix circular import
111 from core.models import ObjectChange
113 exclude = []
114 if get_config().CHANGELOG_SKIP_EMPTY_CHANGES: 114 ↛ 117line 114 didn't jump to line 117 because the condition on line 114 was always true
115 exclude = ['last_updated']
117 objectchange = ObjectChange(
118 changed_object=self,
119 object_repr=str(self)[:200],
120 action=action,
121 message=self._changelog_message or '',
122 )
123 if hasattr(self, '_prechange_snapshot'):
124 objectchange.prechange_data = self._prechange_snapshot
125 if action in (ObjectChangeActionChoices.ACTION_CREATE, ObjectChangeActionChoices.ACTION_UPDATE):
126 self._postchange_snapshot = self.serialize_object(exclude=exclude)
127 objectchange.postchange_data = self._postchange_snapshot
129 return objectchange
130 to_objectchange.alters_data = True
133class CloningMixin(models.Model):
134 """
135 Provides the clone() method used to prepare a copy of existing objects.
136 """
137 class Meta:
138 abstract = True
140 def clone(self):
141 """
142 Returns a dictionary of attributes suitable for creating a copy of the current instance. This is used for pre-
143 populating an object creation form in the UI. By default, this method will replicate any fields listed in the
144 model's `clone_fields` list (if defined), but it can be overridden to apply custom logic.
146 ```python
147 class MyModel(NetBoxModel):
148 def clone(self):
149 attrs = super().clone()
150 attrs['extra-value'] = 123
151 return attrs
152 ```
153 """
154 attrs = {}
156 for field_name in getattr(self, 'clone_fields', []):
157 field = self._meta.get_field(field_name)
159 # A GenericForeignKey is cloned under the subwidget names the creation form's
160 # GenericObjectChoiceField expects (e.g. scope_content_type / scope_object_id).
161 if isinstance(field, GenericForeignKey):
162 content_type_id = getattr(self, f'{field.ct_field}_id', None)
163 object_id = getattr(self, field.fk_field, None)
165 if content_type_id not in (None, '') and object_id not in (None, ''):
166 attrs[f'{field.name}_content_type'] = content_type_id
167 attrs[f'{field.name}_object_id'] = object_id
169 continue
171 field_value = field.value_from_object(self)
173 if field_value and isinstance(field, models.ManyToManyField):
174 attrs[field_name] = [v.pk for v in field_value]
175 elif field_value and isinstance(field, models.JSONField):
176 attrs[field_name] = json.dumps(field_value)
177 elif field_value not in (None, ''):
178 attrs[field_name] = field_value
180 # Include tags (if applicable)
181 if is_taggable(self):
182 attrs['tags'] = [tag.pk for tag in self.tags.all()]
184 # Include any cloneable custom fields
185 if hasattr(self, 'custom_fields'):
186 for field in self.custom_fields:
187 if field.is_cloneable:
188 attrs[f'cf_{field.name}'] = self.custom_field_data.get(field.name)
190 return attrs
193class CustomFieldsMixin(models.Model):
194 """
195 Enables support for custom fields.
196 """
197 custom_field_data = models.JSONField(
198 encoder=CustomFieldJSONEncoder,
199 blank=True,
200 default=dict
201 )
203 class Meta:
204 abstract = True
206 @cached_property
207 def cf(self):
208 """
209 Return a dictionary mapping each custom field for this instance to its deserialized value.
211 ```python
212 >>> tenant = Tenant.objects.first()
213 >>> tenant.cf
214 {'primary_site': <Site: DM-NYC>, 'cust_id': 'DMI01', 'is_active': True}
215 ```
216 """
217 return {
218 cf.name: cf.deserialize(self.custom_field_data.get(cf.name))
219 for cf in self.custom_fields
220 }
222 @cached_property
223 def custom_fields(self):
224 """
225 Return the list of CustomFields assigned to this model.
227 ```python
228 >>> tenant = Tenant.objects.first()
229 >>> tenant.custom_fields
230 [<CustomField: Primary site>, <CustomField: Customer ID>, <CustomField: Is active>]
231 ```
232 """
233 from extras.models import CustomField
234 return CustomField.objects.get_for_model(self)
236 def get_custom_fields(self, omit_hidden=False):
237 """
238 Return a dictionary of custom fields for a single object in the form `{field: value}`.
240 ```python
241 >>> tenant = Tenant.objects.first()
242 >>> tenant.get_custom_fields()
243 {<CustomField: Customer ID>: 'CYB01'}
244 ```
246 Args:
247 omit_hidden: If True, custom fields with no UI visibility will be omitted.
248 """
249 from extras.models import CustomField
250 data = {}
252 for field in CustomField.objects.get_for_model(self):
253 value = self.custom_field_data.get(field.name)
255 # Skip hidden fields if 'omit_hidden' is True
256 if omit_hidden and field.ui_visible == CustomFieldUIVisibleChoices.HIDDEN:
257 continue
258 if omit_hidden and field.ui_visible == CustomFieldUIVisibleChoices.IF_SET and not value:
259 continue
261 data[field] = field.deserialize(value)
263 return data
265 def get_custom_fields_by_group(self):
266 """
267 Return a dictionary of custom field/value mappings organized by group. Hidden fields are omitted.
269 ```python
270 >>> tenant = Tenant.objects.first()
271 >>> tenant.get_custom_fields_by_group()
272 {
273 '': {<CustomField: Primary site>: <Site: DM-NYC>},
274 'Billing': {<CustomField: Customer ID>: 'DMI01', <CustomField: Is active>: True}
275 }
276 ```
277 """
278 from extras.models import CustomField
279 groups = defaultdict(dict)
280 visible_custom_fields = [
281 cf for cf in CustomField.objects.get_for_model(self)
282 if cf.ui_visible != CustomFieldUIVisibleChoices.HIDDEN
283 ]
285 for cf in visible_custom_fields:
286 value = self.custom_field_data.get(cf.name)
287 if value in CUSTOMFIELD_EMPTY_VALUES and cf.ui_visible == CustomFieldUIVisibleChoices.IF_SET:
288 continue
289 value = cf.deserialize(value)
290 groups[cf.group_name][cf] = value
292 return dict(groups)
294 def clean(self):
295 super().clean()
296 from extras.models import CustomField
298 # Fields still being provisioned are fetched alongside the active ones, but are not live:
299 # their stored data belongs to the job acting on it, so it is neither validated below nor
300 # pruned as stale -- while remaining subject to the defaults applied in save(), which draws
301 # on this same set of statuses. Only active fields are validated or enforced as required.
302 assigned_fields = CustomField.objects.get_for_model(
303 self, statuses=CustomFieldStatusChoices.DATA_STATUSES
304 )
305 custom_fields = {
306 cf.name: cf for cf in assigned_fields
307 if cf.status == CustomFieldStatusChoices.STATUS_ACTIVE
308 }
310 # Remove any stale custom field data
311 assigned_names = {cf.name for cf in assigned_fields}
312 self.custom_field_data = {
313 k: v for k, v in self.custom_field_data.items() if k in assigned_names
314 }
316 # Validate all field values
317 for field_name, value in self.custom_field_data.items(): 317 ↛ 318line 317 didn't jump to line 318 because the loop on line 317 never started
318 if (cf := custom_fields.get(field_name)) is None:
319 # The field is not live; its value is left to the job which is provisioning it
320 continue
322 try:
323 cf.validate(value)
324 except ValidationError as e:
325 raise ValidationError(_("Invalid value for custom field '{name}': {error}").format(
326 name=field_name, error=e.message
327 ))
329 # Validate uniqueness if enforced
330 if cf.unique and value not in CUSTOMFIELD_EMPTY_VALUES:
331 if self._meta.model.objects.exclude(pk=self.pk).filter(**{
332 f'custom_field_data__{field_name}': value
333 }).exists():
334 raise ValidationError(_("Custom field '{name}' must have a unique value.").format(
335 name=field_name
336 ))
338 # Check for missing required values
339 for cf in custom_fields.values(): 339 ↛ 340line 339 didn't jump to line 340 because the loop on line 339 never started
340 if cf.required and cf.name not in self.custom_field_data:
341 raise ValidationError(_("Missing required custom field '{name}'.").format(name=cf.name))
343 def save(self, *args, **kwargs):
344 from extras.models import CustomField
346 # Populate default values for custom fields not already present in the object data. This
347 # covers fields still being provisioned as well as active ones, so that an object created
348 # while a new field is being backfilled does not miss its default (see
349 # CustomFieldManager.get_defaults_for_model()).
350 for name, default in CustomField.objects.get_defaults_for_model(self).items(): 350 ↛ 351line 350 didn't jump to line 351 because the loop on line 350 never started
351 if name not in self.custom_field_data:
352 self.custom_field_data[name] = default
354 super().save(*args, **kwargs)
357class CustomLinksMixin(models.Model):
358 """
359 Enables support for custom links.
360 """
361 class Meta:
362 abstract = True
365class CustomValidationMixin(models.Model):
366 """
367 Enables user-configured validation rules for models.
368 """
369 class Meta:
370 abstract = True
372 def clean(self):
373 super().clean()
375 # If the instance is a base for replications, skip custom validation
376 if getattr(self, '_replicated_base', False): 376 ↛ 377line 376 didn't jump to line 377 because the condition on line 376 was never true
377 return
379 # Send the post_clean signal
380 post_clean.send(sender=self.__class__, instance=self)
383class ExportTemplatesMixin(models.Model):
384 """
385 Enables support for export templates.
386 """
387 class Meta:
388 abstract = True
391class ImageAttachmentsMixin(models.Model):
392 """
393 Enables the assignments of ImageAttachments.
394 """
395 images = GenericRelation(
396 to='extras.ImageAttachment',
397 content_type_field='object_type',
398 object_id_field='object_id'
399 )
401 class Meta:
402 abstract = True
405class ContactsMixin(models.Model):
406 """
407 Enables the assignment of Contacts to a model (via ContactAssignment).
408 """
409 contacts = GenericRelation(
410 to='tenancy.ContactAssignment',
411 content_type_field='object_type',
412 object_id_field='object_id'
413 )
415 class Meta:
416 abstract = True
418 def get_contacts(self, inherited=True):
419 """
420 Return a `QuerySet` matching all contacts assigned to this object.
422 Args:
423 inherited: If `True`, inherited contacts from parent objects are included.
424 """
425 from tenancy.models import ContactAssignment
427 from . import NestedGroupModel, NestedLtreeGroupModel
429 filter = Q(
430 object_type=ObjectType.objects.get_for_model(self),
431 object_id__in=(
432 self.get_ancestors(include_self=True)
433 if (isinstance(self, (NestedGroupModel, NestedLtreeGroupModel)) and inherited)
434 else [self.pk]
435 ),
436 )
438 return ContactAssignment.objects.filter(filter)
441class BookmarksMixin(models.Model):
442 """
443 Enables support for user bookmarks.
444 """
445 bookmarks = GenericRelation(
446 to='extras.Bookmark',
447 content_type_field='object_type',
448 object_id_field='object_id'
449 )
451 class Meta:
452 abstract = True
455class NotificationsMixin(models.Model):
456 """
457 Enables support for user notifications.
458 """
459 subscriptions = GenericRelation(
460 to='extras.Subscription',
461 content_type_field='object_type',
462 object_id_field='object_id'
463 )
465 class Meta:
466 abstract = True
469def batch_delete_jobs(job_queryset):
470 """
471 Delete the Jobs in `job_queryset` in JOB_DELETE_BATCH_SIZE chunks. Job cannot be fast-deleted
472 (a global pre_delete receiver forces per-instance signals), so a single delete would build one
473 huge collection of Job instances and run one very long DELETE; batching bounds the per-cycle
474 work. Callers are responsible for wrapping this in a transaction. As with the prior cascade
475 behavior, this bulk delete does not invoke Job.delete() and therefore does not cancel the
476 backing RQ job. See #22812.
477 """
478 from core.models import Job
480 # Route writes to the same database the queryset reads from. In JobsMixin.delete the queryset
481 # is bound to the instance's DB while Job.objects would otherwise use the router default; if
482 # those diverge the deleted rows never leave the read side and the loop below never terminates.
483 jobs = Job.objects.using(job_queryset.db)
485 job_pks = job_queryset.order_by('pk').values_list('pk', flat=True)
486 # Re-slice the queryset each iteration: it re-queries after each batch delete, so the
487 # remaining set shrinks and the loop terminates (do not hoist this into a cursor).
488 while pks := list(job_pks[:JOB_DELETE_BATCH_SIZE]):
489 # only('pk'): the batch still can't fast-delete, so each Job in the batch is instantiated;
490 # loading just the PK avoids pulling the large data/log_entries payloads into memory.
491 jobs.filter(pk__in=pks).only('pk').delete()
494class JobsMixin(models.Model):
495 """
496 Enables support for job results.
498 Note: for the job-batching in delete() to run, JobsMixin must precede DeleteMixin in a
499 model's MRO. DeleteMixin.delete() drives its own collector and does not call super(), so a
500 model declared as e.g. `class Foo(NetBoxModel, JobsMixin)` would reach DeleteMixin first and
501 bypass the batching. Core models that combine both (e.g. DataSource) list JobsMixin first.
502 """
503 jobs = GenericRelation(
504 to='core.Job',
505 content_type_field='object_type',
506 object_id_field='object_id',
507 for_concrete_model=False
508 )
510 class Meta:
511 abstract = True
513 def delete(self, using=None, *args, **kwargs):
514 # Delete associated jobs in batches so the cascade never has to load thousands of Job
515 # rows into memory at once. Wrapped in a transaction so that a failure in the parent
516 # delete rolls the job deletions back as well. See #22812.
517 using = using or router.db_for_write(self.__class__, instance=self)
518 with transaction.atomic(using=using):
519 batch_delete_jobs(self.jobs.using(using))
520 return super().delete(using, *args, **kwargs)
521 delete.alters_data = True
523 def get_latest_jobs(self):
524 """
525 Return a list of the most recent jobs for this instance.
526 """
527 return self.jobs.filter(status__in=JobStatusChoices.TERMINAL_STATE_CHOICES).order_by('-started').defer('data')
530class JournalingMixin(models.Model):
531 """
532 Enables support for object journaling. Adds a generic relation (`journal_entries`)
533 to NetBox's JournalEntry model.
534 """
535 journal_entries = GenericRelation(
536 to='extras.JournalEntry',
537 object_id_field='assigned_object_id',
538 content_type_field='assigned_object_type'
539 )
541 class Meta:
542 abstract = True
545class TagsMixin(models.Model):
546 """
547 Enables support for tag assignment. Assigned tags can be managed via the `tags` attribute,
548 which is a `NetBoxTaggableManager` instance. The field is a `NetBoxTaggableManagerField`,
549 which performs `%(app_label)s` / `%(class)s` interpolation on `related_name` to avoid
550 reverse-accessor collisions between same-named models in different apps (e.g. plugins).
551 """
552 tags = NetBoxTaggableManagerField(
553 through='extras.TaggedItem',
554 ordering=('weight', 'name'),
555 manager=NetBoxTaggableManager,
556 related_name='%(app_label)s_%(class)s_tagged+',
557 )
559 class Meta:
560 abstract = True
563class EventRulesMixin(models.Model):
564 """
565 Enables support for event rules, which can be used to transmit webhooks or execute scripts automatically.
566 """
567 class Meta:
568 abstract = True
571class SyncedDataMixin(models.Model):
572 """
573 Enables population of local data from a DataFile object, synchronized from a remote DataSource.
574 """
575 data_source = models.ForeignKey(
576 to='core.DataSource',
577 on_delete=models.PROTECT,
578 blank=True,
579 null=True,
580 related_name='+',
581 help_text=_("Remote data source")
582 )
583 data_file = models.ForeignKey(
584 to='core.DataFile',
585 on_delete=models.SET_NULL,
586 blank=True,
587 null=True,
588 related_name='+'
589 )
590 data_path = models.CharField(
591 verbose_name=_('data path'),
592 max_length=1000,
593 blank=True,
594 editable=False,
595 help_text=_("Path to remote file (relative to data source root)")
596 )
597 auto_sync_enabled = models.BooleanField(
598 verbose_name=_('auto sync enabled'),
599 default=False,
600 help_text=_("Enable automatic synchronization of data when the data file is updated")
601 )
602 data_synced = models.DateTimeField(
603 verbose_name=_('date synced'),
604 blank=True,
605 null=True,
606 editable=False
607 )
609 class Meta:
610 abstract = True
612 @property
613 def is_synced(self):
614 return self.data_file and self.data_synced >= self.data_file.last_updated
616 def clean(self):
618 if self.data_file: 618 ↛ 619line 618 didn't jump to line 619 because the condition on line 618 was never true
619 self.data_source = self.data_file.source
620 self.data_path = self.data_file.path
621 self.sync()
622 else:
623 self.data_source = None
624 self.data_path = ''
625 self.auto_sync_enabled = False
626 self.data_synced = None
628 super().clean()
629 clean.alters_data = True
631 def save(self, *args, **kwargs):
632 from core.models import AutoSyncRecord
634 ret = super().save(*args, **kwargs)
636 # Create/delete AutoSyncRecord as needed
637 object_type = ObjectType.objects.get_for_model(self)
638 if self.auto_sync_enabled: 638 ↛ 639line 638 didn't jump to line 639 because the condition on line 638 was never true
639 AutoSyncRecord.objects.update_or_create(
640 object_type=object_type,
641 object_id=self.pk,
642 defaults={'datafile': self.data_file}
643 )
644 else:
645 AutoSyncRecord.objects.filter(
646 object_type=object_type,
647 object_id=self.pk
648 ).delete()
650 return ret
652 def delete(self, *args, **kwargs):
653 from core.models import AutoSyncRecord
655 # Delete AutoSyncRecord
656 object_type = ObjectType.objects.get_for_model(self)
657 AutoSyncRecord.objects.filter(
658 object_type=object_type,
659 object_id=self.pk
660 ).delete()
662 return super().delete(*args, **kwargs)
664 def resolve_data_file(self):
665 """
666 Determine the designated DataFile object identified by its parent DataSource and its path. Returns None if
667 either attribute is unset, or if no matching DataFile is found.
668 """
669 from core.models import DataFile
671 if self.data_source and self.data_path:
672 try:
673 return DataFile.objects.get(source=self.data_source, path=self.data_path)
674 except DataFile.DoesNotExist:
675 pass
676 return None
678 def validate_synced_value(self, field_name, value):
679 """
680 Validate content synchronized from the assigned DataFile. Model validation checks fields before clean()
681 runs, so synced values are otherwise never checked.
682 """
683 field = self._meta.get_field(field_name)
684 if value is None and not field.null:
685 raise ValidationError({
686 'data_file': _("The selected data file is empty or its content could not be read.")
687 })
688 try:
689 # Blank and null rules govern user input, so only non-empty content is put through the field
690 if value not in field.empty_values:
691 field.validate(value, self)
692 field.run_validators(value)
693 except ValidationError as e:
694 raise ValidationError({'data_file': e.messages}) from e
696 return value
698 def sync(self, save=False):
699 """
700 Synchronize the object from it's assigned DataFile (if any). This wraps sync_data() and updates
701 the synced_data timestamp.
703 :param save: If true, save() will be called after data has been synchronized
704 """
705 self.sync_data()
706 self.data_synced = timezone.now()
707 if save:
708 self.save()
709 sync.alters_data = True
711 def sync_data(self):
712 """
713 Inheriting models must override this method with specific logic to copy data from the assigned DataFile
714 to the local instance. This method should *NOT* call save() on the instance.
715 """
716 raise NotImplementedError(_("{class_name} must implement a sync_data() method.").format(
717 class_name=self.__class__
718 ))
721#
722# Feature registration
723#
725register_model_feature('bookmarks', lambda model: issubclass(model, BookmarksMixin))
726register_model_feature('change_logging', lambda model: issubclass(model, ChangeLoggingMixin))
727register_model_feature('cloning', lambda model: issubclass(model, CloningMixin))
728register_model_feature('contacts', lambda model: issubclass(model, ContactsMixin))
729register_model_feature('custom_fields', lambda model: issubclass(model, CustomFieldsMixin))
730register_model_feature('custom_links', lambda model: issubclass(model, CustomLinksMixin))
731register_model_feature('custom_validation', lambda model: issubclass(model, CustomValidationMixin))
732register_model_feature('event_rules', lambda model: issubclass(model, EventRulesMixin))
733register_model_feature('export_templates', lambda model: issubclass(model, ExportTemplatesMixin))
734register_model_feature('image_attachments', lambda model: issubclass(model, ImageAttachmentsMixin))
735register_model_feature('jobs', lambda model: issubclass(model, JobsMixin))
736register_model_feature('journaling', lambda model: issubclass(model, JournalingMixin))
737register_model_feature('notifications', lambda model: issubclass(model, NotificationsMixin))
738register_model_feature('synced_data', lambda model: issubclass(model, SyncedDataMixin))
739register_model_feature('tags', lambda model: issubclass(model, TagsMixin))
742def model_is_public(model):
743 """
744 Return True if the model is considered "public use;" otherwise return False.
746 All non-core and non-plugin models are excluded.
747 """
748 opts = model._meta
749 if opts.app_label not in CORE_APPS and not isinstance(opts.app_config, PluginConfig):
750 return False
751 return not getattr(model, '_netbox_private', False)
754def get_model_features(model):
755 """
756 Return all features supported by the given model.
757 """
758 return [
759 feature for feature, test_func in registry['model_features'].items() if test_func(model)
760 ]
763def has_feature(model_or_ct, feature):
764 """
765 Returns True if the model supports the specified feature.
766 """
767 # If an ObjectType was passed, we can use it directly
768 if type(model_or_ct) is ObjectType: 768 ↛ 769line 768 didn't jump to line 769 because the condition on line 768 was never true
769 ot = model_or_ct
770 # If a ContentType was passed, resolve its model class and run the associated feature test
771 elif type(model_or_ct) is ContentType:
772 model = model_or_ct.model_class()
773 if model is None: # Stale content type 773 ↛ 775line 773 didn't jump to line 775 because the condition on line 773 was always true
774 return False
775 try:
776 test_func = registry['model_features'][feature]
777 except KeyError:
778 # Unknown feature
779 return False
780 return test_func(model)
781 # For anything else, look up the ObjectType
782 else:
783 ot = ObjectType.objects.get_for_model(model_or_ct)
784 # ObjectType is invalid/deleted
785 if ot is None: 785 ↛ 786line 785 didn't jump to line 786 because the condition on line 785 was never true
786 return False
787 return feature in ot.features
790def register_models(*models):
791 """
792 Register one or more models in NetBox. This entails:
794 - Determining whether the model is considered "public" (available for reference by other models)
795 - Registering which features the model supports (e.g. bookmarks, custom fields, etc.)
796 - Registering any feature-specific views for the model (e.g. ObjectJournalView instances)
798 register_model() should be called for each relevant model under the ready() of an app's AppConfig class.
799 """
800 from utilities.views import register_model_view
802 for model in models:
803 app_label, model_name = model._meta.label_lower.split('.')
805 # Register applicable feature views for the model
806 if issubclass(model, ContactsMixin):
807 register_model_view(model, 'contacts', kwargs={'model': model})(
808 'netbox.views.generic.ObjectContactsView'
809 )
810 if issubclass(model, JournalingMixin):
811 register_model_view(model, 'journal', kwargs={'model': model})(
812 'netbox.views.generic.ObjectJournalView'
813 )
814 if issubclass(model, ChangeLoggingMixin):
815 register_model_view(model, 'changelog', kwargs={'model': model})(
816 'netbox.views.generic.ObjectChangeLogView'
817 )
818 if issubclass(model, JobsMixin):
819 register_model_view(model, 'jobs', kwargs={'model': model})(
820 'netbox.views.generic.ObjectJobsView'
821 )
822 if issubclass(model, ImageAttachmentsMixin):
823 register_model_view(model, 'image-attachments', kwargs={'model': model})(
824 'netbox.views.generic.ObjectImageAttachmentsView'
825 )
826 if issubclass(model, SyncedDataMixin):
827 register_model_view(model, 'sync', kwargs={'model': model})(
828 'netbox.views.generic.ObjectSyncDataView'
829 )
831 # Auto-register custom permission actions declared in Meta.permissions
832 if meta_permissions := getattr(model._meta, 'permissions', None):
833 actions = [
834 ModelAction(codename, help_text=_(name))
835 for codename, name in meta_permissions
836 ]
837 if actions: 837 ↛ 802line 837 didn't jump to line 802 because the condition on line 837 was always true
838 register_model_actions(model, actions)