features.py 29 KB

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