| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826 |
- import json
- from collections import defaultdict
- from functools import cached_property
- from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
- from django.contrib.contenttypes.models import ContentType
- from django.core.validators import ValidationError
- from django.db import models, router, transaction
- from django.db.models import Q
- from django.utils import timezone
- from django.utils.translation import gettext_lazy as _
- from core.choices import JobStatusChoices, ObjectChangeActionChoices
- from core.models import ObjectType
- from extras.choices import *
- from extras.constants import CUSTOMFIELD_EMPTY_VALUES
- from extras.managers import NetBoxTaggableManager, NetBoxTaggableManagerField
- from extras.utils import is_taggable
- from netbox.config import get_config
- from netbox.constants import CORE_APPS, JOB_DELETE_BATCH_SIZE
- from netbox.models.deletion import DeleteMixin
- from netbox.plugins import PluginConfig
- from netbox.registry import registry
- from netbox.signals import post_clean
- from netbox.utils import register_model_feature
- from utilities.json import CustomFieldJSONEncoder
- from utilities.permissions import ModelAction, register_model_actions
- from utilities.serialization import serialize_object
- __all__ = (
- 'BookmarksMixin',
- 'ChangeLoggingMixin',
- 'CloningMixin',
- 'ContactsMixin',
- 'CustomFieldsMixin',
- 'CustomLinksMixin',
- 'CustomValidationMixin',
- 'EventRulesMixin',
- 'ExportTemplatesMixin',
- 'ImageAttachmentsMixin',
- 'JobsMixin',
- 'JournalingMixin',
- 'NotificationsMixin',
- 'SyncedDataMixin',
- 'TagsMixin',
- 'batch_delete_jobs',
- 'get_model_features',
- 'has_feature',
- 'model_is_public',
- 'register_models',
- )
- #
- # Feature mixins
- #
- class ChangeLoggingMixin(DeleteMixin, models.Model):
- """
- Provides change logging support for a model. Adds the `created` and `last_updated` fields.
- """
- created = models.DateTimeField(
- verbose_name=_('created'),
- auto_now_add=True,
- blank=True,
- null=True
- )
- last_updated = models.DateTimeField(
- verbose_name=_('last updated'),
- auto_now=True,
- blank=True,
- null=True
- )
- class Meta:
- abstract = True
- def __init__(self, *args, **kwargs):
- changelog_message = kwargs.pop('changelog_message', None)
- super().__init__(*args, **kwargs)
- self._changelog_message = changelog_message
- def serialize_object(self, exclude=None):
- """
- Return a JSON representation of the instance. Models can override this method to replace or extend the default
- serialization logic provided by the `serialize_object()` utility function.
- Args:
- exclude: An iterable of attribute names to omit from the serialized output
- """
- return serialize_object(self, exclude=exclude or [])
- def snapshot(self):
- """
- Save a snapshot of the object's current state in preparation for modification. The snapshot is saved as
- `_prechange_snapshot` on the instance.
- """
- exclude_fields = []
- if get_config().CHANGELOG_SKIP_EMPTY_CHANGES:
- exclude_fields = ['last_updated',]
- self._prechange_snapshot = self.serialize_object(exclude=exclude_fields)
- snapshot.alters_data = True
- def to_objectchange(self, action):
- """
- Return a new ObjectChange representing a change made to this object. This will typically be called automatically
- by ChangeLoggingMiddleware.
- """
- # TODO: Fix circular import
- from core.models import ObjectChange
- exclude = []
- if get_config().CHANGELOG_SKIP_EMPTY_CHANGES:
- exclude = ['last_updated']
- objectchange = ObjectChange(
- changed_object=self,
- object_repr=str(self)[:200],
- action=action,
- message=self._changelog_message or '',
- )
- if hasattr(self, '_prechange_snapshot'):
- objectchange.prechange_data = self._prechange_snapshot
- if action in (ObjectChangeActionChoices.ACTION_CREATE, ObjectChangeActionChoices.ACTION_UPDATE):
- self._postchange_snapshot = self.serialize_object(exclude=exclude)
- objectchange.postchange_data = self._postchange_snapshot
- return objectchange
- to_objectchange.alters_data = True
- class CloningMixin(models.Model):
- """
- Provides the clone() method used to prepare a copy of existing objects.
- """
- class Meta:
- abstract = True
- def clone(self):
- """
- Returns a dictionary of attributes suitable for creating a copy of the current instance. This is used for pre-
- populating an object creation form in the UI. By default, this method will replicate any fields listed in the
- model's `clone_fields` list (if defined), but it can be overridden to apply custom logic.
- ```python
- class MyModel(NetBoxModel):
- def clone(self):
- attrs = super().clone()
- attrs['extra-value'] = 123
- return attrs
- ```
- """
- attrs = {}
- for field_name in getattr(self, 'clone_fields', []):
- field = self._meta.get_field(field_name)
- # A GenericForeignKey is cloned under the subwidget names the creation form's
- # GenericObjectChoiceField expects (e.g. scope_content_type / scope_object_id).
- if isinstance(field, GenericForeignKey):
- content_type_id = getattr(self, f'{field.ct_field}_id', None)
- object_id = getattr(self, field.fk_field, None)
- if content_type_id not in (None, '') and object_id not in (None, ''):
- attrs[f'{field.name}_content_type'] = content_type_id
- attrs[f'{field.name}_object_id'] = object_id
- continue
- field_value = field.value_from_object(self)
- if field_value and isinstance(field, models.ManyToManyField):
- attrs[field_name] = [v.pk for v in field_value]
- elif field_value and isinstance(field, models.JSONField):
- attrs[field_name] = json.dumps(field_value)
- elif field_value not in (None, ''):
- attrs[field_name] = field_value
- # Include tags (if applicable)
- if is_taggable(self):
- attrs['tags'] = [tag.pk for tag in self.tags.all()]
- # Include any cloneable custom fields
- if hasattr(self, 'custom_fields'):
- for field in self.custom_fields:
- if field.is_cloneable:
- attrs[f'cf_{field.name}'] = self.custom_field_data.get(field.name)
- return attrs
- class CustomFieldsMixin(models.Model):
- """
- Enables support for custom fields.
- """
- custom_field_data = models.JSONField(
- encoder=CustomFieldJSONEncoder,
- blank=True,
- default=dict
- )
- class Meta:
- abstract = True
- @cached_property
- def cf(self):
- """
- Return a dictionary mapping each custom field for this instance to its deserialized value.
- ```python
- >>> tenant = Tenant.objects.first()
- >>> tenant.cf
- {'primary_site': <Site: DM-NYC>, 'cust_id': 'DMI01', 'is_active': True}
- ```
- """
- return {
- cf.name: cf.deserialize(self.custom_field_data.get(cf.name))
- for cf in self.custom_fields
- }
- @cached_property
- def custom_fields(self):
- """
- Return the list of CustomFields assigned to this model.
- ```python
- >>> tenant = Tenant.objects.first()
- >>> tenant.custom_fields
- [<CustomField: Primary site>, <CustomField: Customer ID>, <CustomField: Is active>]
- ```
- """
- from extras.models import CustomField
- return CustomField.objects.get_for_model(self)
- def get_custom_fields(self, omit_hidden=False):
- """
- Return a dictionary of custom fields for a single object in the form `{field: value}`.
- ```python
- >>> tenant = Tenant.objects.first()
- >>> tenant.get_custom_fields()
- {<CustomField: Customer ID>: 'CYB01'}
- ```
- Args:
- omit_hidden: If True, custom fields with no UI visibility will be omitted.
- """
- from extras.models import CustomField
- data = {}
- for field in CustomField.objects.get_for_model(self):
- value = self.custom_field_data.get(field.name)
- # Skip hidden fields if 'omit_hidden' is True
- if omit_hidden and field.ui_visible == CustomFieldUIVisibleChoices.HIDDEN:
- continue
- if omit_hidden and field.ui_visible == CustomFieldUIVisibleChoices.IF_SET and not value:
- continue
- data[field] = field.deserialize(value)
- return data
- def get_custom_fields_by_group(self):
- """
- Return a dictionary of custom field/value mappings organized by group. Hidden fields are omitted.
- ```python
- >>> tenant = Tenant.objects.first()
- >>> tenant.get_custom_fields_by_group()
- {
- '': {<CustomField: Primary site>: <Site: DM-NYC>},
- 'Billing': {<CustomField: Customer ID>: 'DMI01', <CustomField: Is active>: True}
- }
- ```
- """
- from extras.models import CustomField
- groups = defaultdict(dict)
- visible_custom_fields = [
- cf for cf in CustomField.objects.get_for_model(self)
- if cf.ui_visible != CustomFieldUIVisibleChoices.HIDDEN
- ]
- for cf in visible_custom_fields:
- value = self.custom_field_data.get(cf.name)
- if value in CUSTOMFIELD_EMPTY_VALUES and cf.ui_visible == CustomFieldUIVisibleChoices.IF_SET:
- continue
- value = cf.deserialize(value)
- groups[cf.group_name][cf] = value
- return dict(groups)
- def populate_custom_field_defaults(self):
- """
- Apply the default value for each custom field
- """
- for cf in self.custom_fields:
- self.custom_field_data[cf.name] = cf.default
- populate_custom_field_defaults.alters_data = True
- def clean(self):
- super().clean()
- from extras.models import CustomField
- # Fields still being provisioned are fetched alongside the active ones, but are not live:
- # their stored data belongs to the job acting on it, so it is neither validated below nor
- # pruned as stale -- while remaining subject to the defaults applied in save(), which draws
- # on this same set of statuses. Only active fields are validated or enforced as required.
- assigned_fields = CustomField.objects.get_for_model(
- self, statuses=CustomFieldStatusChoices.DATA_STATUSES
- )
- custom_fields = {
- cf.name: cf for cf in assigned_fields
- if cf.status == CustomFieldStatusChoices.STATUS_ACTIVE
- }
- # Remove any stale custom field data
- assigned_names = {cf.name for cf in assigned_fields}
- self.custom_field_data = {
- k: v for k, v in self.custom_field_data.items() if k in assigned_names
- }
- # Validate all field values
- for field_name, value in self.custom_field_data.items():
- if (cf := custom_fields.get(field_name)) is None:
- # The field is not live; its value is left to the job which is provisioning it
- continue
- try:
- cf.validate(value)
- except ValidationError as e:
- raise ValidationError(_("Invalid value for custom field '{name}': {error}").format(
- name=field_name, error=e.message
- ))
- # Validate uniqueness if enforced
- if cf.unique and value not in CUSTOMFIELD_EMPTY_VALUES:
- if self._meta.model.objects.exclude(pk=self.pk).filter(**{
- f'custom_field_data__{field_name}': value
- }).exists():
- raise ValidationError(_("Custom field '{name}' must have a unique value.").format(
- name=field_name
- ))
- # Check for missing required values
- for cf in custom_fields.values():
- if cf.required and cf.name not in self.custom_field_data:
- raise ValidationError(_("Missing required custom field '{name}'.").format(name=cf.name))
- def save(self, *args, **kwargs):
- from extras.models import CustomField
- # Populate default values for custom fields not already present in the object data. This
- # covers fields still being provisioned as well as active ones, so that an object created
- # while a new field is being backfilled does not miss its default (see
- # CustomFieldManager.get_defaults_for_model()).
- for name, default in CustomField.objects.get_defaults_for_model(self).items():
- if name not in self.custom_field_data:
- self.custom_field_data[name] = default
- super().save(*args, **kwargs)
- class CustomLinksMixin(models.Model):
- """
- Enables support for custom links.
- """
- class Meta:
- abstract = True
- class CustomValidationMixin(models.Model):
- """
- Enables user-configured validation rules for models.
- """
- class Meta:
- abstract = True
- def clean(self):
- super().clean()
- # If the instance is a base for replications, skip custom validation
- if getattr(self, '_replicated_base', False):
- return
- # Send the post_clean signal
- post_clean.send(sender=self.__class__, instance=self)
- class ExportTemplatesMixin(models.Model):
- """
- Enables support for export templates.
- """
- class Meta:
- abstract = True
- class ImageAttachmentsMixin(models.Model):
- """
- Enables the assignments of ImageAttachments.
- """
- images = GenericRelation(
- to='extras.ImageAttachment',
- content_type_field='object_type',
- object_id_field='object_id'
- )
- class Meta:
- abstract = True
- class ContactsMixin(models.Model):
- """
- Enables the assignment of Contacts to a model (via ContactAssignment).
- """
- contacts = GenericRelation(
- to='tenancy.ContactAssignment',
- content_type_field='object_type',
- object_id_field='object_id'
- )
- class Meta:
- abstract = True
- def get_contacts(self, inherited=True):
- """
- Return a `QuerySet` matching all contacts assigned to this object.
- Args:
- inherited: If `True`, inherited contacts from parent objects are included.
- """
- from tenancy.models import ContactAssignment
- from . import NestedGroupModel, NestedLtreeGroupModel
- filter = Q(
- object_type=ObjectType.objects.get_for_model(self),
- object_id__in=(
- self.get_ancestors(include_self=True)
- if (isinstance(self, (NestedGroupModel, NestedLtreeGroupModel)) and inherited)
- else [self.pk]
- ),
- )
- return ContactAssignment.objects.filter(filter)
- class BookmarksMixin(models.Model):
- """
- Enables support for user bookmarks.
- """
- bookmarks = GenericRelation(
- to='extras.Bookmark',
- content_type_field='object_type',
- object_id_field='object_id'
- )
- class Meta:
- abstract = True
- class NotificationsMixin(models.Model):
- """
- Enables support for user notifications.
- """
- subscriptions = GenericRelation(
- to='extras.Subscription',
- content_type_field='object_type',
- object_id_field='object_id'
- )
- class Meta:
- abstract = True
- def batch_delete_jobs(job_queryset):
- """
- Delete the Jobs in `job_queryset` in JOB_DELETE_BATCH_SIZE chunks. Job cannot be fast-deleted
- (a global pre_delete receiver forces per-instance signals), so a single delete would build one
- huge collection of Job instances and run one very long DELETE; batching bounds the per-cycle
- work. Callers are responsible for wrapping this in a transaction. As with the prior cascade
- behavior, this bulk delete does not invoke Job.delete() and therefore does not cancel the
- backing RQ job. See #22812.
- """
- from core.models import Job
- # Route writes to the same database the queryset reads from. In JobsMixin.delete the queryset
- # is bound to the instance's DB while Job.objects would otherwise use the router default; if
- # those diverge the deleted rows never leave the read side and the loop below never terminates.
- jobs = Job.objects.using(job_queryset.db)
- job_pks = job_queryset.order_by('pk').values_list('pk', flat=True)
- # Re-slice the queryset each iteration: it re-queries after each batch delete, so the
- # remaining set shrinks and the loop terminates (do not hoist this into a cursor).
- while pks := list(job_pks[:JOB_DELETE_BATCH_SIZE]):
- # only('pk'): the batch still can't fast-delete, so each Job in the batch is instantiated;
- # loading just the PK avoids pulling the large data/log_entries payloads into memory.
- jobs.filter(pk__in=pks).only('pk').delete()
- class JobsMixin(models.Model):
- """
- Enables support for job results.
- Note: for the job-batching in delete() to run, JobsMixin must precede DeleteMixin in a
- model's MRO. DeleteMixin.delete() drives its own collector and does not call super(), so a
- model declared as e.g. `class Foo(NetBoxModel, JobsMixin)` would reach DeleteMixin first and
- bypass the batching. Core models that combine both (e.g. DataSource) list JobsMixin first.
- """
- jobs = GenericRelation(
- to='core.Job',
- content_type_field='object_type',
- object_id_field='object_id',
- for_concrete_model=False
- )
- class Meta:
- abstract = True
- def delete(self, using=None, *args, **kwargs):
- # Delete associated jobs in batches so the cascade never has to load thousands of Job
- # rows into memory at once. Wrapped in a transaction so that a failure in the parent
- # delete rolls the job deletions back as well. See #22812.
- using = using or router.db_for_write(self.__class__, instance=self)
- with transaction.atomic(using=using):
- batch_delete_jobs(self.jobs.using(using))
- return super().delete(using, *args, **kwargs)
- delete.alters_data = True
- def get_latest_jobs(self):
- """
- Return a list of the most recent jobs for this instance.
- """
- return self.jobs.filter(status__in=JobStatusChoices.TERMINAL_STATE_CHOICES).order_by('-started').defer('data')
- class JournalingMixin(models.Model):
- """
- Enables support for object journaling. Adds a generic relation (`journal_entries`)
- to NetBox's JournalEntry model.
- """
- journal_entries = GenericRelation(
- to='extras.JournalEntry',
- object_id_field='assigned_object_id',
- content_type_field='assigned_object_type'
- )
- class Meta:
- abstract = True
- class TagsMixin(models.Model):
- """
- Enables support for tag assignment. Assigned tags can be managed via the `tags` attribute,
- which is a `NetBoxTaggableManager` instance. The field is a `NetBoxTaggableManagerField`,
- which performs `%(app_label)s` / `%(class)s` interpolation on `related_name` to avoid
- reverse-accessor collisions between same-named models in different apps (e.g. plugins).
- """
- tags = NetBoxTaggableManagerField(
- through='extras.TaggedItem',
- ordering=('weight', 'name'),
- manager=NetBoxTaggableManager,
- related_name='%(app_label)s_%(class)s_tagged+',
- )
- class Meta:
- abstract = True
- class EventRulesMixin(models.Model):
- """
- Enables support for event rules, which can be used to transmit webhooks or execute scripts automatically.
- """
- class Meta:
- abstract = True
- class SyncedDataMixin(models.Model):
- """
- Enables population of local data from a DataFile object, synchronized from a remote DataSource.
- """
- data_source = models.ForeignKey(
- to='core.DataSource',
- on_delete=models.PROTECT,
- blank=True,
- null=True,
- related_name='+',
- help_text=_("Remote data source")
- )
- data_file = models.ForeignKey(
- to='core.DataFile',
- on_delete=models.SET_NULL,
- blank=True,
- null=True,
- related_name='+'
- )
- data_path = models.CharField(
- verbose_name=_('data path'),
- max_length=1000,
- blank=True,
- editable=False,
- help_text=_("Path to remote file (relative to data source root)")
- )
- auto_sync_enabled = models.BooleanField(
- verbose_name=_('auto sync enabled'),
- default=False,
- help_text=_("Enable automatic synchronization of data when the data file is updated")
- )
- data_synced = models.DateTimeField(
- verbose_name=_('date synced'),
- blank=True,
- null=True,
- editable=False
- )
- class Meta:
- abstract = True
- @property
- def is_synced(self):
- return self.data_file and self.data_synced >= self.data_file.last_updated
- def clean(self):
- if self.data_file:
- self.data_source = self.data_file.source
- self.data_path = self.data_file.path
- self.sync()
- else:
- self.data_source = None
- self.data_path = ''
- self.auto_sync_enabled = False
- self.data_synced = None
- super().clean()
- clean.alters_data = True
- def save(self, *args, **kwargs):
- from core.models import AutoSyncRecord
- ret = super().save(*args, **kwargs)
- # Create/delete AutoSyncRecord as needed
- object_type = ObjectType.objects.get_for_model(self)
- if self.auto_sync_enabled:
- AutoSyncRecord.objects.update_or_create(
- object_type=object_type,
- object_id=self.pk,
- defaults={'datafile': self.data_file}
- )
- else:
- AutoSyncRecord.objects.filter(
- object_type=object_type,
- object_id=self.pk
- ).delete()
- return ret
- def delete(self, *args, **kwargs):
- from core.models import AutoSyncRecord
- # Delete AutoSyncRecord
- object_type = ObjectType.objects.get_for_model(self)
- AutoSyncRecord.objects.filter(
- object_type=object_type,
- object_id=self.pk
- ).delete()
- return super().delete(*args, **kwargs)
- def resolve_data_file(self):
- """
- Determine the designated DataFile object identified by its parent DataSource and its path. Returns None if
- either attribute is unset, or if no matching DataFile is found.
- """
- from core.models import DataFile
- if self.data_source and self.data_path:
- try:
- return DataFile.objects.get(source=self.data_source, path=self.data_path)
- except DataFile.DoesNotExist:
- pass
- return None
- def sync(self, save=False):
- """
- Synchronize the object from it's assigned DataFile (if any). This wraps sync_data() and updates
- the synced_data timestamp.
- :param save: If true, save() will be called after data has been synchronized
- """
- self.sync_data()
- self.data_synced = timezone.now()
- if save:
- self.save()
- sync.alters_data = True
- def sync_data(self):
- """
- Inheriting models must override this method with specific logic to copy data from the assigned DataFile
- to the local instance. This method should *NOT* call save() on the instance.
- """
- raise NotImplementedError(_("{class_name} must implement a sync_data() method.").format(
- class_name=self.__class__
- ))
- #
- # Feature registration
- #
- register_model_feature('bookmarks', lambda model: issubclass(model, BookmarksMixin))
- register_model_feature('change_logging', lambda model: issubclass(model, ChangeLoggingMixin))
- register_model_feature('cloning', lambda model: issubclass(model, CloningMixin))
- register_model_feature('contacts', lambda model: issubclass(model, ContactsMixin))
- register_model_feature('custom_fields', lambda model: issubclass(model, CustomFieldsMixin))
- register_model_feature('custom_links', lambda model: issubclass(model, CustomLinksMixin))
- register_model_feature('custom_validation', lambda model: issubclass(model, CustomValidationMixin))
- register_model_feature('event_rules', lambda model: issubclass(model, EventRulesMixin))
- register_model_feature('export_templates', lambda model: issubclass(model, ExportTemplatesMixin))
- register_model_feature('image_attachments', lambda model: issubclass(model, ImageAttachmentsMixin))
- register_model_feature('jobs', lambda model: issubclass(model, JobsMixin))
- register_model_feature('journaling', lambda model: issubclass(model, JournalingMixin))
- register_model_feature('notifications', lambda model: issubclass(model, NotificationsMixin))
- register_model_feature('synced_data', lambda model: issubclass(model, SyncedDataMixin))
- register_model_feature('tags', lambda model: issubclass(model, TagsMixin))
- 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 all features supported by the given model.
- """
- return [
- feature for feature, test_func in registry['model_features'].items() if test_func(model)
- ]
- 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 and run the associated feature test
- elif type(model_or_ct) is ContentType:
- model = model_or_ct.model_class()
- if model is None: # Stale content type
- return False
- try:
- test_func = registry['model_features'][feature]
- except KeyError:
- # Unknown feature
- return False
- return test_func(model)
- # For anything else, look up the ObjectType
- else:
- ot = ObjectType.objects.get_for_model(model_or_ct)
- # ObjectType is invalid/deleted
- if ot is None:
- return False
- return feature in ot.features
- def register_models(*models):
- """
- Register one or more models in NetBox. This entails:
- - Determining whether the model is considered "public" (available for reference by other models)
- - Registering which features the model supports (e.g. bookmarks, custom fields, etc.)
- - Registering any feature-specific views for the model (e.g. ObjectJournalView instances)
- register_model() should be called for each relevant model under the ready() of an app's AppConfig class.
- """
- from utilities.views import register_model_view
- for model in models:
- app_label, model_name = model._meta.label_lower.split('.')
- # Register applicable feature views for the model
- if issubclass(model, ContactsMixin):
- register_model_view(model, 'contacts', kwargs={'model': model})(
- 'netbox.views.generic.ObjectContactsView'
- )
- if issubclass(model, JournalingMixin):
- register_model_view(model, 'journal', kwargs={'model': model})(
- 'netbox.views.generic.ObjectJournalView'
- )
- if issubclass(model, ChangeLoggingMixin):
- register_model_view(model, 'changelog', kwargs={'model': model})(
- 'netbox.views.generic.ObjectChangeLogView'
- )
- if issubclass(model, JobsMixin):
- register_model_view(model, 'jobs', kwargs={'model': model})(
- 'netbox.views.generic.ObjectJobsView'
- )
- if issubclass(model, ImageAttachmentsMixin):
- register_model_view(model, 'image-attachments', kwargs={'model': model})(
- 'netbox.views.generic.ObjectImageAttachmentsView'
- )
- if issubclass(model, SyncedDataMixin):
- register_model_view(model, 'sync', kwargs={'model': model})(
- 'netbox.views.generic.ObjectSyncDataView'
- )
- # Auto-register custom permission actions declared in Meta.permissions
- if meta_permissions := getattr(model._meta, 'permissions', None):
- actions = [
- ModelAction(codename, help_text=_(name))
- for codename, name in meta_permissions
- ]
- if actions:
- register_model_actions(model, actions)
|