Coverage for extras/models/models.py: 52%
428 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import json
2import re
3import urllib.parse
4from pathlib import Path
6from django.conf import settings
7from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
8from django.contrib.postgres.fields import ArrayField
9from django.core.exceptions import ValidationError
10from django.core.validators import MaxValueValidator, MinValueValidator
11from django.db import models
12from django.urls import reverse
13from django.utils import timezone
14from django.utils.html import escape
15from django.utils.safestring import mark_safe
16from django.utils.text import format_lazy
17from django.utils.translation import gettext_lazy as _
18from rest_framework.utils.encoders import JSONEncoder
20from extras.choices import *
21from extras.conditions import ConditionSet, InvalidCondition
22from extras.constants import *
23from extras.models.mixins import RenderTemplateMixin
24from extras.querysets import SharedObjectQuerySet
25from extras.utils import image_upload
26from netbox.config import get_config
27from netbox.event_rules import get_event_rule_action, get_event_rule_action_choices
28from netbox.events import get_event_type_choices
29from netbox.models import ChangeLoggedModel
30from netbox.models.features import (
31 CloningMixin,
32 CustomFieldsMixin,
33 CustomLinksMixin,
34 ExportTemplatesMixin,
35 SyncedDataMixin,
36 TagsMixin,
37 has_feature,
38)
39from netbox.models.mixins import OwnerMixin
40from netbox.settings_utils import parse_job_timeout
41from utilities.html import clean_html
42from utilities.jinja2 import JINJA2_TEMPLATE_RE, render_jinja2, sanitize_http_header, validate_jinja2_syntax
43from utilities.querydict import dict_to_querydict
44from utilities.querysets import RestrictedQuerySet
45from utilities.tables import get_table_for_model
47__all__ = (
48 'Bookmark',
49 'CustomLink',
50 'EventRule',
51 'ExportTemplate',
52 'ImageAttachment',
53 'JournalEntry',
54 'SavedFilter',
55 'TableConfig',
56 'Webhook',
57)
59# Matches a literal URL scheme (RFC 3986), independent of urlsplit()'s netloc parsing -- which can
60# raise ValueError on a malformed host -- so a payload_url's scheme can always be read even when
61# its host is templated or malformed.
62LITERAL_SCHEME_RE = re.compile(r'^([a-zA-Z][a-zA-Z0-9+.-]*):')
65class EventRule(CustomFieldsMixin, ExportTemplatesMixin, OwnerMixin, TagsMixin, ChangeLoggedModel):
66 """
67 An EventRule defines an action to be taken automatically in response to a specific set of events, such as when a
68 specific type of object is created, modified, or deleted. The action to be taken might entail transmitting a
69 webhook or executing a custom script.
70 """
71 object_types = models.ManyToManyField(
72 to='contenttypes.ContentType',
73 related_name='event_rules',
74 verbose_name=_('object types'),
75 help_text=_("The object(s) to which this rule applies.")
76 )
77 name = models.CharField(
78 verbose_name=_('name'),
79 max_length=150,
80 unique=True
81 )
82 description = models.CharField(
83 verbose_name=_('description'),
84 max_length=200,
85 blank=True
86 )
87 event_types = ArrayField(
88 base_field=models.CharField(max_length=50, choices=get_event_type_choices),
89 help_text=_("The types of event which will trigger this rule.")
90 )
91 enabled = models.BooleanField(
92 verbose_name=_('enabled'),
93 default=True
94 )
95 conditions = models.JSONField(
96 verbose_name=_('conditions'),
97 blank=True,
98 null=True,
99 help_text=_("A set of conditions which determine whether the event will be generated.")
100 )
102 # Action to take
103 action_type = models.CharField(
104 max_length=100,
105 # Bare callable, re-evaluated fresh on each access via Django's CallableChoiceIterator,
106 # so a plugin action registered after this module was first imported is still reflected.
107 choices=get_event_rule_action_choices,
108 default=EventRuleActionChoices.WEBHOOK,
109 verbose_name=_('action type')
110 )
111 action_object_type = models.ForeignKey(
112 to='contenttypes.ContentType',
113 related_name='eventrule_actions',
114 on_delete=models.CASCADE,
115 blank=True,
116 null=True,
117 )
118 action_object_id = models.PositiveBigIntegerField(
119 blank=True,
120 null=True
121 )
122 action_object = GenericForeignKey(
123 ct_field='action_object_type',
124 fk_field='action_object_id'
125 )
126 action_data = models.JSONField(
127 verbose_name=_('data'),
128 blank=True,
129 null=True,
130 help_text=_("Additional data to pass to the action object")
131 )
132 comments = models.TextField(
133 verbose_name=_('comments'),
134 blank=True
135 )
137 class Meta:
138 ordering = ('name',)
139 indexes = (
140 models.Index(fields=('action_object_type', 'action_object_id')),
141 )
142 verbose_name = _('event rule')
143 verbose_name_plural = _('event rules')
145 def __str__(self):
146 return self.name
148 def get_absolute_url(self):
149 return reverse('extras:eventrule', args=[self.pk])
151 @property
152 def action_provider(self):
153 """
154 Return the registered EventRuleAction instance for this rule's action_type, or None if it
155 is not currently registered (e.g. the providing plugin is not installed).
156 """
157 return get_event_rule_action(self.action_type)
159 @property
160 def action_is_available(self):
161 return self.action_provider is not None
163 def get_action_type_display(self):
164 if action := self.action_provider:
165 return action.label
166 return _('{slug} (unavailable)').format(slug=self.action_type)
168 def get_action_type_color(self):
169 return None if self.action_is_available else 'red'
171 def clean(self):
172 super().clean()
174 # Validate that any conditions are in the correct format
175 if self.conditions:
176 try:
177 ConditionSet(self.conditions)
178 except ValueError as e:
179 raise ValidationError({'conditions': e})
181 # action_data must be a JSON object (or null)
182 if self.action_data is not None and not isinstance(self.action_data, dict):
183 raise ValidationError({'action_data': _('Action data must be a JSON object or null.')})
185 # action_type's own validity is already enforced by the field's dynamic choices= (Field.
186 # validate(), earlier in full_clean()); guard here only in case clean() ran standalone.
187 if self.action_is_available:
188 self.action_provider._validate(action_object=self.action_object, action_data=self.action_data)
190 def eval_conditions(self, data):
191 """
192 Test whether the given data meets the conditions of the event rule (if any). Return True
193 if met or no conditions are specified.
194 """
195 if not self.conditions:
196 return True
198 logger = logging.getLogger('netbox.event_rules')
200 try:
201 result = ConditionSet(self.conditions).eval(data)
202 logger.debug(f'{self.name}: Evaluated as {result}')
203 return result
204 except InvalidCondition as e:
205 logger.error(f"{self.name}: Evaluation failed. {e}")
206 return False
209class Webhook(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, OwnerMixin, ChangeLoggedModel):
210 """
211 A Webhook defines a request that will be sent to a remote application when an object is created, updated, and/or
212 delete in NetBox. The request will contain a representation of the object, which the remote application can act on.
213 Each Webhook can be limited to firing only on certain actions or certain object types.
214 """
215 name = models.CharField(
216 verbose_name=_('name'),
217 max_length=150,
218 unique=True
219 )
220 description = models.CharField(
221 verbose_name=_('description'),
222 max_length=200,
223 blank=True
224 )
225 payload_url = models.CharField(
226 max_length=500,
227 verbose_name=_('URL'),
228 help_text=_(
229 "This URL will be called using the HTTP method defined when the webhook is called. Must be "
230 "http:// or https://. Jinja2 template processing is supported (with the same context as the "
231 "request body) for part or all of the URL."
232 )
233 )
234 http_method = models.CharField(
235 max_length=30,
236 choices=WebhookHttpMethodChoices,
237 default=WebhookHttpMethodChoices.METHOD_POST,
238 verbose_name=_('HTTP method')
239 )
240 http_content_type = models.CharField(
241 max_length=100,
242 default=HTTP_CONTENT_TYPE_JSON,
243 verbose_name=_('HTTP content type'),
244 help_text=_(
245 'The complete list of official content types is available '
246 '<a href="https://www.iana.org/assignments/media-types/media-types.xhtml">here</a>.'
247 )
248 )
249 additional_headers = models.TextField(
250 verbose_name=_('additional headers'),
251 blank=True,
252 help_text=_(
253 "User-supplied HTTP headers to be sent with the request in addition to the HTTP content type. Headers "
254 "should be defined in the format <code>Name: Value</code>. Jinja2 template processing is supported with "
255 "the same context as the request body (below). When interpolating untrusted data (such as object "
256 "attributes) into a header value, apply the <code>header_safe</code> filter to guard against HTTP header "
257 "injection, e.g. <code>X-Object: {{ data.name | header_safe }}</code>."
258 )
259 )
260 body_template = models.TextField(
261 verbose_name=_('body template'),
262 blank=True,
263 help_text=_(
264 "Jinja2 template for a custom request body. If blank, a JSON object representing the change will be "
265 "included. Available context data includes: <code>event</code>, <code>model</code>, "
266 "<code>timestamp</code>, <code>request</code>, and <code>data</code>."
267 )
268 )
269 secret = models.CharField(
270 verbose_name=_('secret'),
271 max_length=255,
272 blank=True,
273 help_text=_(
274 "When provided, the request will include a <code>X-Hook-Signature</code> header containing a HMAC hex "
275 "digest of the payload body using the secret as the key. The secret is not transmitted in the request."
276 )
277 )
278 ssl_verification = models.BooleanField(
279 default=True,
280 verbose_name=_('SSL verification'),
281 help_text=_("Enable SSL certificate verification. Disable with caution!")
282 )
283 ca_file_path = models.CharField(
284 max_length=4096,
285 null=True,
286 blank=True,
287 verbose_name=_('CA File Path'),
288 help_text=_(
289 "The specific CA certificate file to use for SSL verification. Leave blank to use the system defaults."
290 )
291 )
292 timeout = models.PositiveSmallIntegerField(
293 verbose_name=_('timeout'),
294 null=True,
295 blank=True,
296 validators=(
297 MinValueValidator(1),
298 MaxValueValidator(3600),
299 ),
300 help_text=format_lazy(
301 _(
302 "The maximum time (in seconds) to wait for a response before failing the request. Leave blank to use "
303 "the system default ({default_timeout} seconds)."
304 ),
305 default_timeout=settings.WEBHOOK_DEFAULT_TIMEOUT
306 )
307 )
308 events = GenericRelation(
309 EventRule,
310 content_type_field='action_object_type',
311 object_id_field='action_object_id'
312 )
314 class Meta:
315 ordering = ('name',)
316 verbose_name = _('webhook')
317 verbose_name_plural = _('webhooks')
319 def __str__(self):
320 return self.name
322 def get_absolute_url(self):
323 return reverse('extras:webhook', args=[self.pk])
325 @property
326 def docs_url(self):
327 return f'{settings.STATIC_URL}docs/models/extras/webhook/'
329 def clean(self):
330 super().clean()
332 errors = {}
334 # CA file path requires SSL verification enabled
335 if not self.ssl_verification and self.ca_file_path: 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true
336 errors['ca_file_path'] = _('Do not specify a CA certificate file if SSL verification is disabled.')
338 # payload_url may be a literal URL or a Jinja2 template (see its help_text). Skipped when
339 # blank; clean_fields() already flags that.
340 if self.payload_url: 340 ↛ 364line 340 didn't jump to line 364 because the condition on line 340 was always true
341 if JINJA2_TEMPLATE_RE.search(self.payload_url): 341 ↛ 345line 341 didn't jump to line 345 because the condition on line 341 was never true
342 # A literal, disallowed scheme (e.g. "file://") can never resolve no matter what
343 # else in the value is templated; anything else is checked for template syntax
344 # only, since its rendered result isn't known here.
345 match = LITERAL_SCHEME_RE.match(self.payload_url)
346 if match and match.group(1).lower() not in ('http', 'https'):
347 errors['payload_url'] = _("Enter a valid URL, beginning with http:// or https://.")
348 else:
349 try:
350 validate_jinja2_syntax(self.payload_url)
351 except ValidationError as e:
352 errors['payload_url'] = e
353 else:
354 # Fully literal -- validate directly rather than via URLValidator, which rejects
355 # single-label and underscore hosts that `requests` accepts fine. urlsplit() can
356 # raise ValueError for a malformed netloc (e.g. an unbalanced IPv6 bracket).
357 try:
358 scheme, netloc = urllib.parse.urlsplit(self.payload_url)[:2]
359 except ValueError:
360 scheme, netloc = '', ''
361 if scheme not in ('http', 'https') or not netloc: 361 ↛ 364line 361 didn't jump to line 364 because the condition on line 361 was always true
362 errors['payload_url'] = _("Enter a valid URL, beginning with http:// or https://.")
364 if errors: 364 ↛ 370line 364 didn't jump to line 370 because the condition on line 364 was always true
365 raise ValidationError(errors)
367 # A timeout which meets or exceeds the background job timeout leaves no room for the request's own timeout
368 # to apply: the worker will terminate the job first. (Staying below the job timeout does not guarantee that
369 # the request times out on its own, as the timeout applies separately to connecting and to reading data.)
370 job_timeout = parse_job_timeout(settings.RQ_DEFAULT_TIMEOUT)
371 if self.timeout is not None and job_timeout is not None and self.timeout >= job_timeout:
372 raise ValidationError({
373 'timeout': _(
374 "Timeout must be less than the background job timeout ({timeout} seconds)."
375 ).format(timeout=job_timeout)
376 })
378 def render_headers(self, context):
379 """
380 Render additional_headers and return a dict of Header: Value pairs.
381 """
382 if not self.additional_headers:
383 return {}
384 ret = {}
385 # Expose the `header_safe` filter so template authors can sanitize interpolated values (e.g. user-controlled
386 # object data) against HTTP header (CR/LF) injection. See utilities.jinja2.sanitize_http_header.
387 data = render_jinja2(self.additional_headers, context, filters={'header_safe': sanitize_http_header})
388 for line in data.splitlines():
389 if ':' not in line:
390 continue
391 header, value = line.split(':', 1)
392 ret[header.strip()] = value.strip()
393 return ret
395 def render_body(self, context):
396 """
397 Render the body template, if defined. Otherwise, jump the context as a JSON object.
398 """
399 if self.body_template:
400 return render_jinja2(self.body_template, context)
401 return json.dumps(context, cls=JSONEncoder)
403 def render_payload_url(self, context):
404 """
405 Render the payload URL.
406 """
407 return render_jinja2(self.payload_url, context)
410class CustomLink(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedModel):
411 """
412 A custom link to an external representation of a NetBox object. The link text and URL fields accept Jinja2 template
413 code to be rendered with an object as context.
414 """
415 object_types = models.ManyToManyField(
416 to='contenttypes.ContentType',
417 related_name='custom_links',
418 help_text=_('The object type(s) to which this link applies.')
419 )
420 name = models.CharField(
421 verbose_name=_('name'),
422 max_length=100,
423 unique=True
424 )
425 enabled = models.BooleanField(
426 verbose_name=_('enabled'),
427 default=True
428 )
429 link_text = models.TextField(
430 verbose_name=_('link text'),
431 help_text=_("Jinja2 template code for link text")
432 )
433 link_url = models.TextField(
434 verbose_name=_('link URL'),
435 help_text=_("Jinja2 template code for link URL")
436 )
437 weight = models.PositiveSmallIntegerField(
438 verbose_name=_('weight'),
439 default=100
440 )
441 group_name = models.CharField(
442 verbose_name=_('group name'),
443 max_length=50,
444 blank=True,
445 help_text=_("Links with the same group will appear as a dropdown menu")
446 )
447 button_class = models.CharField(
448 verbose_name=_('button class'),
449 max_length=30,
450 choices=CustomLinkButtonClassChoices,
451 default=CustomLinkButtonClassChoices.DEFAULT,
452 help_text=_("The class of the first link in a group will be used for the dropdown button")
453 )
454 new_window = models.BooleanField(
455 verbose_name=_('new window'),
456 default=False,
457 help_text=_("Force link to open in a new window")
458 )
460 clone_fields = (
461 'object_types', 'enabled', 'weight', 'group_name', 'button_class', 'new_window',
462 )
464 class Meta:
465 ordering = ['group_name', 'weight', 'name']
466 indexes = (
467 models.Index(fields=('group_name', 'weight', 'name')), # Default ordering
468 )
469 verbose_name = _('custom link')
470 verbose_name_plural = _('custom links')
472 def __str__(self):
473 return self.name
475 def get_absolute_url(self):
476 return reverse('extras:customlink', args=[self.pk])
478 @property
479 def docs_url(self):
480 return f'{settings.STATIC_URL}docs/models/extras/customlink/'
482 def render(self, context):
483 """
484 Render the CustomLink given the provided context, and return the text, link, and link_target.
486 :param context: The context passed to Jinja2
487 """
488 text = render_jinja2(self.link_text, context).strip()
489 if not text:
490 return {}
491 link = render_jinja2(self.link_url, context).strip()
492 link_target = ' target="_blank"' if self.new_window else ''
494 # Sanitize link text
495 allowed_schemes = get_config().ALLOWED_URL_SCHEMES
496 text = clean_html(text, allowed_schemes)
498 # Sanitize link
499 link = urllib.parse.quote(link, safe='/:?&=%+[]@#,;!')
501 # Verify link scheme is allowed
502 result = urllib.parse.urlparse(link)
503 if result.scheme and result.scheme not in allowed_schemes:
504 link = ""
506 return {
507 'text': text,
508 'link': link,
509 'link_target': link_target,
510 }
513class ExportTemplate(
514 SyncedDataMixin,
515 CloningMixin,
516 ExportTemplatesMixin,
517 OwnerMixin,
518 ChangeLoggedModel,
519 RenderTemplateMixin,
520):
521 object_types = models.ManyToManyField(
522 to='contenttypes.ContentType',
523 related_name='export_templates',
524 help_text=_('The object type(s) to which this template applies.')
525 )
526 name = models.CharField(
527 verbose_name=_('name'),
528 max_length=100
529 )
530 description = models.CharField(
531 verbose_name=_('description'),
532 max_length=200,
533 blank=True
534 )
536 clone_fields = (
537 'object_types', 'template_code', 'mime_type', 'file_name', 'file_extension', 'as_attachment',
538 )
540 class Meta:
541 ordering = ('name',)
542 indexes = (
543 models.Index(fields=('name',)), # Default ordering
544 )
545 verbose_name = _('export template')
546 verbose_name_plural = _('export templates')
548 def __str__(self):
549 return self.name
551 def get_absolute_url(self):
552 return reverse('extras:exporttemplate', args=[self.pk])
554 @property
555 def docs_url(self):
556 return f'{settings.STATIC_URL}docs/models/extras/exporttemplate/'
558 def clean(self):
559 super().clean()
561 if self.name.lower() == 'table':
562 raise ValidationError({
563 'name': _('"{name}" is a reserved name. Please choose a different name.').format(name=self.name)
564 })
566 def sync_data(self):
567 """
568 Synchronize template content from the designated DataFile (if any).
569 """
570 self.template_code = self.validate_synced_value('template_code', self.data_file.data_as_string)
571 sync_data.alters_data = True
573 def get_context(self, context=None, queryset=None):
574 _context = super().get_context(context=context, queryset=queryset)
575 _context['queryset'] = queryset
576 return _context
579class SavedFilter(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedModel):
580 """
581 A set of predefined keyword parameters that can be reused to filter for specific objects.
582 """
583 object_types = models.ManyToManyField(
584 to='contenttypes.ContentType',
585 related_name='saved_filters',
586 help_text=_('The object type(s) to which this filter applies.')
587 )
588 name = models.CharField(
589 verbose_name=_('name'),
590 max_length=100,
591 unique=True
592 )
593 slug = models.SlugField(
594 verbose_name=_('slug'),
595 max_length=100,
596 unique=True
597 )
598 description = models.CharField(
599 verbose_name=_('description'),
600 max_length=200,
601 blank=True
602 )
603 user = models.ForeignKey(
604 to=settings.AUTH_USER_MODEL,
605 on_delete=models.SET_NULL,
606 blank=True,
607 null=True
608 )
609 weight = models.PositiveSmallIntegerField(
610 verbose_name=_('weight'),
611 default=100
612 )
613 enabled = models.BooleanField(
614 verbose_name=_('enabled'),
615 default=True
616 )
617 shared = models.BooleanField(
618 verbose_name=_('shared'),
619 default=True
620 )
621 parameters = models.JSONField(
622 verbose_name=_('parameters')
623 )
625 objects = SharedObjectQuerySet.as_manager()
627 clone_fields = (
628 'object_types', 'weight', 'enabled', 'parameters',
629 )
631 class Meta:
632 ordering = ('weight', 'name')
633 indexes = (
634 models.Index(fields=('weight', 'name')), # Default ordering
635 )
636 verbose_name = _('saved filter')
637 verbose_name_plural = _('saved filters')
639 def __str__(self):
640 return self.name
642 def get_absolute_url(self):
643 return reverse('extras:savedfilter', args=[self.pk])
645 @property
646 def docs_url(self):
647 return f'{settings.STATIC_URL}docs/models/extras/savedfilter/'
649 def clean(self):
650 super().clean()
652 # Verify that `parameters` is a JSON object
653 if type(self.parameters) is not dict:
654 raise ValidationError(
655 {'parameters': _('Filter parameters must be stored as a dictionary of keyword arguments.')}
656 )
658 @property
659 def url_params(self):
660 qd = dict_to_querydict(self.parameters)
661 return qd.urlencode()
664class TableConfig(CloningMixin, ChangeLoggedModel):
665 """
666 A saved configuration of columns and ordering which applies to a specific table.
667 """
668 object_type = models.ForeignKey(
669 to='contenttypes.ContentType',
670 on_delete=models.CASCADE,
671 related_name='table_configs',
672 help_text=_("The table's object type"),
673 )
674 table = models.CharField(
675 verbose_name=_('table'),
676 max_length=100,
677 )
678 name = models.CharField(
679 verbose_name=_('name'),
680 max_length=100,
681 )
682 description = models.CharField(
683 verbose_name=_('description'),
684 max_length=200,
685 blank=True,
686 )
687 user = models.ForeignKey(
688 to=settings.AUTH_USER_MODEL,
689 on_delete=models.SET_NULL,
690 blank=True,
691 null=True,
692 )
693 weight = models.PositiveSmallIntegerField(
694 verbose_name=_('weight'),
695 default=1000,
696 )
697 enabled = models.BooleanField(
698 verbose_name=_('enabled'),
699 default=True
700 )
701 shared = models.BooleanField(
702 verbose_name=_('shared'),
703 default=True
704 )
705 columns = ArrayField(
706 base_field=models.CharField(max_length=100),
707 )
708 ordering = ArrayField(
709 base_field=models.CharField(max_length=100),
710 blank=True,
711 null=True,
712 )
714 objects = SharedObjectQuerySet.as_manager()
716 clone_fields = ('object_type', 'table', 'enabled', 'shared', 'columns', 'ordering')
718 class Meta:
719 ordering = ('weight', 'name')
720 indexes = (
721 models.Index(fields=('weight', 'name')), # Default ordering
722 )
723 verbose_name = _('table config')
724 verbose_name_plural = _('table configs')
726 def __str__(self):
727 return self.name
729 def get_absolute_url(self):
730 return reverse('extras:tableconfig', args=[self.pk])
732 @property
733 def docs_url(self):
734 return f'{settings.STATIC_URL}docs/models/extras/tableconfig/'
736 @property
737 def table_class(self):
738 return get_table_for_model(self.object_type.model_class(), name=self.table)
740 @property
741 def ordering_items(self):
742 """
743 Return a list of two-tuples indicating the column(s) by which the table is to be ordered and a boolean for each
744 column indicating whether its ordering is ascending.
745 """
746 items = []
747 for col in self.ordering or []:
748 if col.startswith('-'):
749 ascending = False
750 col = col[1:]
751 else:
752 ascending = True
753 items.append((col, ascending))
754 return items
756 def clean(self):
757 super().clean()
759 # Skip table validation until the object type and table have been set
760 if not self.object_type_id or not self.table:
761 return
763 # Validate table
764 if self.table_class is None:
765 raise ValidationError({
766 'table': _("Unknown table: {name}").format(name=self.table)
767 })
769 table = self.table_class([])
771 # Validate ordering columns
772 for name in self.ordering or []:
773 if name.startswith('-'):
774 name = name[1:] # Strip leading hyphen
775 if name not in table.columns:
776 raise ValidationError({
777 'ordering': _('Unknown column: {name}').format(name=name)
778 })
780 # Validate selected columns
781 for name in self.columns or []:
782 if name not in table.columns:
783 raise ValidationError({
784 'columns': _('Unknown column: {name}').format(name=name)
785 })
788class ImageAttachment(ChangeLoggedModel):
789 """
790 An uploaded image which is associated with an object.
791 """
792 object_type = models.ForeignKey(
793 to='contenttypes.ContentType',
794 on_delete=models.CASCADE
795 )
796 object_id = models.PositiveBigIntegerField()
797 parent = GenericForeignKey(
798 ct_field='object_type',
799 fk_field='object_id'
800 )
801 image = models.ImageField(
802 upload_to=image_upload,
803 height_field='image_height',
804 width_field='image_width'
805 )
806 image_height = models.PositiveSmallIntegerField(
807 verbose_name=_('image height'),
808 )
809 image_width = models.PositiveSmallIntegerField(
810 verbose_name=_('image width'),
811 )
812 # Unlike image_height/image_width (populated automatically by ImageField), there is no native size_field, so
813 # this is populated in save(). It is nullable because existing rows predate the field and storage reads can
814 # fail; a null value means "not yet computed" and the size property falls back to reading storage.
815 image_size = models.PositiveBigIntegerField(
816 verbose_name=_('image size'),
817 blank=True,
818 null=True,
819 )
820 name = models.CharField(
821 verbose_name=_('name'),
822 max_length=50,
823 blank=True
824 )
825 description = models.CharField(
826 verbose_name=_('description'),
827 max_length=200,
828 blank=True
829 )
831 objects = RestrictedQuerySet.as_manager()
833 def __init__(self, *args, **kwargs):
834 super().__init__(*args, **kwargs)
836 # Cache an identity for the current image so save() can detect a new/replaced file and recompute the cached
837 # image_size. We combine the file name with the (auto-populated) dimensions: a replacement that reuses the
838 # same name is still caught when its dimensions differ. Read the raw image value from __dict__ to avoid
839 # triggering the ImageField descriptor here (doing so during ORM/GraphQL instantiation can recurse).
840 self._orig_image_key = self._image_identity()
842 def _image_identity(self):
843 """
844 Return a tuple identifying the current image file for change detection: its name plus the dimensions Django
845 populates from it. All three are read raw from __dict__ to avoid triggering the ImageField descriptor
846 (accessing `self.image` during ORM/GraphQL instantiation can recurse). Not a content fingerprint: a
847 replacement with an identical name AND identical dimensions is not distinguished (would require reading the
848 file, the storage round-trip this caching avoids).
849 """
850 original = self.__dict__.get('image')
851 name = getattr(original, 'name', original)
852 return (name, self.__dict__.get('image_height'), self.__dict__.get('image_width'))
854 class Meta:
855 ordering = ('name', 'pk') # name may be non-unique
856 indexes = (
857 models.Index(fields=('name', 'id')), # Default ordering
858 models.Index(fields=('object_type', 'object_id')),
859 )
860 verbose_name = _('image attachment')
861 verbose_name_plural = _('image attachments')
863 def __str__(self):
864 return self.name or self.filename
866 def get_absolute_url(self):
867 return reverse('extras:imageattachment', args=[self.pk])
869 def clean(self):
870 super().clean()
872 # Validate the assigned object type
873 if not has_feature(self.object_type, 'image_attachments'):
874 raise ValidationError(
875 _("Image attachments cannot be assigned to this object type ({type}).").format(type=self.object_type)
876 )
878 def delete(self, *args, **kwargs):
880 _name = self.image.name
882 super().delete(*args, **kwargs)
884 # Delete file from disk
885 self.image.delete(save=False)
887 # Deleting the file erases its name. We restore the image's filename here in case we still need to reference it
888 # before the request finishes. (For example, to display a message indicating the ImageAttachment was deleted.)
889 self.image.name = _name
891 @property
892 def filename(self):
893 base_name = Path(self.image.name).name
894 prefix = f"{self.object_type.model}_{self.object_id}_"
895 return base_name.removeprefix(prefix)
897 @property
898 def html_tag(self):
899 """
900 Returns a complete <img> tag suitable for embedding in an HTML document.
901 """
902 return mark_safe('<img src="{url}" height="{height}" width="{width}" alt="{alt_text}" />'.format(
903 url=self.image.url,
904 height=self.image_height,
905 width=self.image_width,
906 alt_text=escape(self.description or self.name),
907 ))
909 def _read_image_size(self):
910 """
911 Read the image file's size from storage, suppressing an OSError in case the file is inaccessible. Also
912 opportunistically catch other exceptions that we know other storage back-ends to throw. Returns None if the
913 size cannot be determined. This may issue a request to the storage backend (e.g. a HEAD request to S3).
914 """
915 if not self.image:
916 return None
918 expected_exceptions = [OSError]
920 try:
921 from botocore.exceptions import ClientError
922 expected_exceptions.append(ClientError)
923 except ImportError:
924 pass
926 try:
927 return self.image.size
928 except tuple(expected_exceptions):
929 return None
931 @property
932 def size(self):
933 """
934 Return the size of the image file in bytes. Prefer the cached `image_size` value to avoid a storage request;
935 fall back to reading from storage for legacy rows where `image_size` has not yet been populated.
936 """
937 if self.image_size is not None:
938 return self.image_size
939 return self._read_image_size()
941 def save(self, *args, **kwargs):
942 # Populate image_size on creation or when the image file has changed. Reading the size may touch the storage
943 # backend (e.g. a HEAD request to S3), so we only do it when necessary: bulk operations that don't alter the
944 # image (bulk edit, rename) leave the identity unchanged and skip the read entirely. We never overwrite a good
945 # value with None (e.g. on a transient storage error); a failed read while replacing a file keeps the prior
946 # size until the next successful save, which is preferred over storing None.
947 orig_image_key = getattr(self, '_orig_image_key', None)
948 if self._state.adding or self._image_identity() != orig_image_key:
949 size = self._read_image_size()
950 if size is not None:
951 self.image_size = size
953 super().save(*args, **kwargs)
955 # Refresh the cached identity so subsequent saves on this instance detect further changes correctly.
956 self._orig_image_key = self._image_identity()
958 def to_objectchange(self, action):
959 objectchange = super().to_objectchange(action)
960 objectchange.related_object = self.parent
961 return objectchange
964class JournalEntry(CustomFieldsMixin, CustomLinksMixin, TagsMixin, ExportTemplatesMixin, ChangeLoggedModel):
965 """
966 A historical remark concerning an object; collectively, these form an object's journal. The journal is used to
967 preserve historical context around an object, and complements NetBox's built-in change logging. For example, you
968 might record a new journal entry when a device undergoes maintenance, or when a prefix is expanded.
969 """
970 assigned_object_type = models.ForeignKey(
971 to='contenttypes.ContentType',
972 on_delete=models.CASCADE
973 )
974 assigned_object_id = models.PositiveBigIntegerField()
975 assigned_object = GenericForeignKey(
976 ct_field='assigned_object_type',
977 fk_field='assigned_object_id'
978 )
979 created_by = models.ForeignKey(
980 to=settings.AUTH_USER_MODEL,
981 on_delete=models.SET_NULL,
982 blank=True,
983 null=True
984 )
985 kind = models.CharField(
986 verbose_name=_('kind'),
987 max_length=30,
988 choices=JournalEntryKindChoices,
989 default=JournalEntryKindChoices.KIND_INFO
990 )
991 comments = models.TextField(
992 verbose_name=_('comments'),
993 )
995 class Meta:
996 ordering = ('-created',)
997 indexes = (
998 models.Index(fields=('-created',)), # Default ordering
999 models.Index(fields=('assigned_object_type', 'assigned_object_id')),
1000 )
1001 verbose_name = _('journal entry')
1002 verbose_name_plural = _('journal entries')
1004 def __str__(self):
1005 created = timezone.localtime(self.created)
1006 return (
1007 f"{created.date().isoformat()} {created.time().isoformat(timespec='minutes')} "
1008 f"({self.get_kind_display()})"
1009 )
1011 def get_absolute_url(self):
1012 return reverse('extras:journalentry', args=[self.pk])
1014 def clean(self):
1015 super().clean()
1017 # Validate the assigned object type
1018 if not has_feature(self.assigned_object_type, 'journaling'):
1019 raise ValidationError(
1020 _("Journaling is not supported for this object type ({type}).").format(type=self.assigned_object_type)
1021 )
1023 def get_kind_color(self):
1024 return JournalEntryKindChoices.colors.get(self.kind)
1027class Bookmark(models.Model):
1028 """
1029 An object bookmarked by a User.
1030 """
1031 created = models.DateTimeField(
1032 verbose_name=_('created'),
1033 auto_now_add=True
1034 )
1035 object_type = models.ForeignKey(
1036 to='contenttypes.ContentType',
1037 on_delete=models.PROTECT
1038 )
1039 object_id = models.PositiveBigIntegerField()
1040 object = GenericForeignKey(
1041 ct_field='object_type',
1042 fk_field='object_id'
1043 )
1044 user = models.ForeignKey(
1045 to=settings.AUTH_USER_MODEL,
1046 on_delete=models.CASCADE
1047 )
1049 objects = RestrictedQuerySet.as_manager()
1051 class Meta:
1052 ordering = ('created', 'pk')
1053 indexes = (
1054 models.Index(fields=('created', 'id')), # Default ordering
1055 models.Index(fields=('object_type', 'object_id')),
1056 )
1057 constraints = (
1058 models.UniqueConstraint(
1059 fields=('object_type', 'object_id', 'user'),
1060 name='%(app_label)s_%(class)s_unique_per_object_and_user'
1061 ),
1062 )
1063 verbose_name = _('bookmark')
1064 verbose_name_plural = _('bookmarks')
1066 def __str__(self):
1067 if self.object:
1068 return str(self.object)
1069 return super().__str__()
1071 def get_absolute_url(self):
1072 return reverse('account:bookmarks')
1074 def clean(self):
1075 super().clean()
1077 # Validate the assigned object type
1078 if not has_feature(self.object_type, 'bookmarks'):
1079 raise ValidationError(
1080 _("Bookmarks cannot be assigned to this object type ({type}).").format(type=self.object_type)
1081 )