Coverage for extras/models/customfields.py: 18%
559 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import copy
2import decimal
3import json
4import re
5from datetime import date, datetime
7import django_filters
8import jsonschema
9from django import forms
10from django.conf import settings
11from django.core.validators import RegexValidator, ValidationError
12from django.db import connections, models, router, transaction
13from django.db.models import F, Func, Q, Value
14from django.urls import reverse
15from django.utils.html import escape
16from django.utils.safestring import mark_safe
17from django.utils.translation import gettext_lazy as _
18from jsonschema.exceptions import ValidationError as JSONValidationError
20from core.models import ObjectType
21from extras.choices import *
22from extras.data import CHOICE_SETS
23from extras.fields import ChoiceSetField
24from netbox.constants import ADVISORY_LOCK_KEYS
25from netbox.context import query_cache
26from netbox.models import ChangeLoggedModel
27from netbox.models.features import CloningMixin, ExportTemplatesMixin
28from netbox.models.mixins import OwnerMixin
29from netbox.search import FieldTypes
30from utilities import filters
31from utilities.datetime import datetime_from_timestamp
32from utilities.exceptions import AbortRequest
33from utilities.forms.fields import (
34 CSVChoiceField,
35 CSVModelChoiceField,
36 CSVModelMultipleChoiceField,
37 CSVMultipleChoiceField,
38 DynamicChoiceField,
39 DynamicModelChoiceField,
40 DynamicModelMultipleChoiceField,
41 DynamicMultipleChoiceField,
42 JSONField,
43 LaxURLField,
44)
45from utilities.forms.utils import add_blank_choice
46from utilities.forms.widgets import APISelect, APISelectMultiple, DatePicker, DateTimePicker
47from utilities.jsonschema import validate_schema
48from utilities.querysets import RestrictedQuerySet, chunked_update
49from utilities.templatetags.builtins.filters import render_markdown
50from utilities.validators import url_scheme_is_allowed, validate_regex
52__all__ = (
53 'CustomField',
54 'CustomFieldChoiceSet',
55 'CustomFieldManager',
56)
58SEARCH_TYPES = {
59 CustomFieldTypeChoices.TYPE_TEXT: FieldTypes.STRING,
60 CustomFieldTypeChoices.TYPE_LONGTEXT: FieldTypes.STRING,
61 CustomFieldTypeChoices.TYPE_INTEGER: FieldTypes.INTEGER,
62 CustomFieldTypeChoices.TYPE_DECIMAL: FieldTypes.FLOAT,
63 CustomFieldTypeChoices.TYPE_DATE: FieldTypes.STRING,
64 CustomFieldTypeChoices.TYPE_URL: FieldTypes.STRING,
65}
68class CustomFieldManager(models.Manager.from_queryset(RestrictedQuerySet)):
69 use_in_migrations = True
71 def get_for_model(self, model, statuses=(CustomFieldStatusChoices.STATUS_ACTIVE,)):
72 """
73 Return a list of the CustomFields assigned to the given model which hold one of the given
74 statuses.
76 Only active fields are returned by default: a field awaiting a bulk update of its stored data
77 is not live, and must be invisible to every consumer of custom field data until that work
78 completes (see CustomFieldStatusChoices). This is the sole entry point by which custom fields
79 are resolved for an object, so excluding them here excludes them everywhere.
81 Every assigned field is fetched and cached whichever statuses are asked for, so that callers
82 wanting different subsets share one query per model per request.
84 Args:
85 model: The model whose custom fields are to be returned
86 statuses: The statuses to select (active only by default)
87 """
88 cache = query_cache.get()
90 # Check the request cache before hitting the database. Test the cached value against None
91 # rather than for truthiness: a model with no custom fields caches an empty list, which
92 # would otherwise be treated as a miss and re-queried on every call.
93 custom_fields = cache['custom_fields'].get(model._meta.model) if cache is not None else None
94 if custom_fields is None:
95 content_type = ObjectType.objects.get_for_model(model._meta.concrete_model)
96 custom_fields = list(
97 self.get_queryset().filter(object_types=content_type).select_related(
98 'related_object_type', 'choice_set'
99 )
100 )
102 # Populate the request cache to avoid redundant lookups
103 if cache is not None:
104 cache['custom_fields'][model._meta.model] = custom_fields
106 return [cf for cf in custom_fields if cf.status in statuses]
108 def get_defaults_for_model(self, model):
109 """
110 Return a dictionary of serialized default values for all CustomFields applicable to the given model.
112 Fields still being provisioned are included, unlike in get_for_model(). The provisioning job
113 backfills only the objects which predate the field, so an object created while it runs must
114 pick up the default here or never receive one at all.
116 The defaults are assembled on each call from the fields cached by get_for_model() rather than
117 cached in their own right: building them costs a pass over a handful of objects already in
118 memory, where a second cache would have to be kept coherent with the first.
119 """
120 custom_fields = self.get_for_model(model, statuses=CustomFieldStatusChoices.DATA_STATUSES)
122 # Copied so that a mutable default cannot be aliased into the object data of every object
123 # which takes it, the fields above being cached for the life of the request.
124 return {
125 cf.name: copy.deepcopy(cf.default) for cf in custom_fields if cf.default is not None
126 }
128 @staticmethod
129 def clear_cache():
130 """
131 Discard the custom fields cached for the current request, so that a subsequent read reflects
132 a change which has been applied to the database without passing through save().
134 Called wherever a field's status is written directly (see CustomFieldStatusChoices): the
135 cache spans the whole of a request -- and the whole of a script or job run -- so a field
136 taken offline, brought live, or marked for deletion partway through one would otherwise
137 remain visible, or invisible, to everything which followed it there.
138 """
139 if (cache := query_cache.get()) is not None:
140 cache['custom_fields'].clear()
143class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedModel):
144 object_types = models.ManyToManyField(
145 to='contenttypes.ContentType',
146 related_name='custom_fields',
147 help_text=_('The object(s) to which this field applies.')
148 )
149 type = models.CharField(
150 verbose_name=_('type'),
151 max_length=50,
152 choices=CustomFieldTypeChoices,
153 default=CustomFieldTypeChoices.TYPE_TEXT,
154 help_text=_('The type of data this custom field holds')
155 )
156 related_object_type = models.ForeignKey(
157 to='contenttypes.ContentType',
158 on_delete=models.PROTECT,
159 blank=True,
160 null=True,
161 help_text=_('The type of NetBox object this field maps to (for object fields)')
162 )
163 name = models.CharField(
164 verbose_name=_('name'),
165 max_length=50,
166 unique=True,
167 help_text=_('Internal field name'),
168 validators=(
169 RegexValidator(
170 regex=r'^[a-z0-9_]+$',
171 message=_("Only alphanumeric characters and underscores are allowed."),
172 flags=re.IGNORECASE
173 ),
174 RegexValidator(
175 regex=r'__',
176 message=_("Double underscores are not permitted in custom field names."),
177 flags=re.IGNORECASE,
178 inverse_match=True
179 ),
180 )
181 )
182 status = models.CharField(
183 max_length=50,
184 choices=CustomFieldStatusChoices,
185 default=CustomFieldStatusChoices.STATUS_ACTIVE,
186 verbose_name=_('status'),
187 help_text=_("Operational state of the field"),
188 editable=False
189 )
190 label = models.CharField(
191 verbose_name=_('label'),
192 max_length=50,
193 blank=True,
194 help_text=_(
195 "Name of the field as displayed to users (if not provided, 'the field's name will be used)"
196 )
197 )
198 group_name = models.CharField(
199 verbose_name=_('group name'),
200 max_length=50,
201 blank=True,
202 help_text=_("Custom fields within the same group will be displayed together")
203 )
204 description = models.CharField(
205 verbose_name=_('description'),
206 max_length=200,
207 blank=True
208 )
209 required = models.BooleanField(
210 verbose_name=_('required'),
211 default=False,
212 help_text=_("This field is required when creating new objects or editing an existing object.")
213 )
214 unique = models.BooleanField(
215 verbose_name=_('must be unique'),
216 default=False,
217 help_text=_("The value of this field must be unique for the assigned object")
218 )
219 search_weight = models.PositiveSmallIntegerField(
220 verbose_name=_('search weight'),
221 default=1000,
222 help_text=_(
223 "Weighting for search. Lower values are considered more important. Fields with a search weight of zero "
224 "will be ignored."
225 )
226 )
227 filter_logic = models.CharField(
228 verbose_name=_('filter logic'),
229 max_length=50,
230 choices=CustomFieldFilterLogicChoices,
231 default=CustomFieldFilterLogicChoices.FILTER_LOOSE,
232 help_text=_("Loose matches any instance of a given string; exact matches the entire field.")
233 )
234 default = models.JSONField(
235 verbose_name=_('default'),
236 blank=True,
237 null=True,
238 help_text=_(
239 'Default value for the field (must be a JSON value). Encapsulate strings with double quotes (e.g. "Foo").'
240 )
241 )
242 related_object_filter = models.JSONField(
243 blank=True,
244 null=True,
245 help_text=_(
246 'Filter the object selection choices using a query_params dict (must be a JSON value).'
247 'Encapsulate strings with double quotes (e.g. "Foo").'
248 )
249 )
250 weight = models.PositiveSmallIntegerField(
251 default=100,
252 verbose_name=_('display weight'),
253 help_text=_('Fields with higher weights appear lower in a form.')
254 )
255 validation_minimum = models.DecimalField(
256 max_digits=16,
257 decimal_places=4,
258 blank=True,
259 null=True,
260 verbose_name=_('minimum value'),
261 help_text=_('Minimum allowed value (for numeric fields)')
262 )
263 validation_maximum = models.DecimalField(
264 max_digits=16,
265 decimal_places=4,
266 blank=True,
267 null=True,
268 verbose_name=_('maximum value'),
269 help_text=_('Maximum allowed value (for numeric fields)')
270 )
271 validation_regex = models.CharField(
272 blank=True,
273 validators=[validate_regex],
274 max_length=500,
275 verbose_name=_('validation regex'),
276 help_text=_(
277 'Regular expression to enforce on text field values. Use ^ and $ to force matching of entire string. For '
278 'example, <code>^[A-Z]{3}$</code> will limit values to exactly three uppercase letters.'
279 )
280 )
281 validation_schema = models.JSONField(
282 blank=True,
283 null=True,
284 validators=[validate_schema],
285 verbose_name=_('validation schema'),
286 help_text=_('A JSON schema definition for validating the custom field value')
287 )
288 choice_set = models.ForeignKey(
289 to='CustomFieldChoiceSet',
290 on_delete=models.PROTECT,
291 related_name='choices_for',
292 verbose_name=_('choice set'),
293 blank=True,
294 null=True
295 )
296 ui_visible = models.CharField(
297 max_length=50,
298 choices=CustomFieldUIVisibleChoices,
299 default=CustomFieldUIVisibleChoices.ALWAYS,
300 verbose_name=_('UI visible'),
301 help_text=_('Specifies whether the custom field is displayed in the UI')
302 )
303 ui_editable = models.CharField(
304 max_length=50,
305 choices=CustomFieldUIEditableChoices,
306 default=CustomFieldUIEditableChoices.YES,
307 verbose_name=_('UI editable'),
308 help_text=_('Specifies whether the custom field value can be edited in the UI')
309 )
310 is_cloneable = models.BooleanField(
311 default=False,
312 verbose_name=_('is cloneable'),
313 help_text=_('Replicate this value when cloning objects')
314 )
315 nulls_first = models.BooleanField(
316 default=True,
317 verbose_name=_('nulls first'),
318 help_text=_('Sort null values before non-null values when ordering by this field')
319 )
320 comments = models.TextField(
321 verbose_name=_('comments'),
322 blank=True
323 )
325 objects = CustomFieldManager()
327 clone_fields = (
328 'object_types', 'type', 'related_object_type', 'group_name', 'description', 'required', 'unique',
329 'search_weight', 'filter_logic', 'default', 'weight', 'validation_minimum', 'validation_maximum',
330 'validation_regex', 'validation_schema', 'choice_set', 'ui_visible', 'ui_editable', 'is_cloneable',
331 'nulls_first',
332 )
334 class Meta:
335 ordering = ['group_name', 'weight', 'name']
336 indexes = (
337 models.Index(fields=('group_name', 'weight', 'name')), # Default ordering
338 )
339 verbose_name = _('custom field')
340 verbose_name_plural = _('custom fields')
342 def __str__(self):
343 return self.label or self.name.replace('_', ' ').capitalize()
345 def get_absolute_url(self):
346 return reverse('extras:customfield', args=[self.pk])
348 @property
349 def docs_url(self):
350 return f'{settings.STATIC_URL}docs/models/extras/customfield/'
352 def __init__(self, *args, **kwargs):
353 super().__init__(*args, **kwargs)
355 # Cache instance's original name so we can check later whether it has changed
356 self._name = self.__dict__.get('name')
358 @property
359 def search_type(self):
360 return SEARCH_TYPES.get(self.type)
362 @property
363 def choices(self):
364 if self.choice_set:
365 return self.choice_set.choices
366 return []
368 def get_status_color(self):
369 return CustomFieldStatusChoices.colors.get(self.status)
371 def get_ui_visible_color(self):
372 return CustomFieldUIVisibleChoices.colors.get(self.ui_visible)
374 def get_ui_editable_color(self):
375 return CustomFieldUIEditableChoices.colors.get(self.ui_editable)
377 def get_choice_label(self, value):
378 if not hasattr(self, '_choice_map'):
379 self._choice_map = dict(self.choices)
380 return self._choice_map.get(value, value)
382 def get_choice_color(self, value):
383 if self.choice_set:
384 return self.choice_set.get_choice_color(value)
385 return None
387 def resolve_selection_value(self, value):
388 """
389 For a Selection or Multiple selection field, wrap the value(s) with their resolved label as
390 {'value': ..., 'label': ...} (a list thereof for multi-select). Other field types pass through
391 unchanged. Shared by the REST API and GraphQL so selection labels resolve consistently (#20897).
392 """
393 if value is None:
394 return value
395 if self.type == CustomFieldTypeChoices.TYPE_SELECT:
396 return {'value': value, 'label': self.get_choice_label(value)}
397 if self.type == CustomFieldTypeChoices.TYPE_MULTISELECT:
398 return [{'value': v, 'label': self.get_choice_label(v)} for v in value]
399 return value
401 @staticmethod
402 def data_lock_key(pk):
403 """
404 The advisory lock which serializes bulk updates of a field's stored data against one another
405 and against its deletion, keyed by primary key so that work on one field never waits on
406 another.
407 """
408 return ADVISORY_LOCK_KEYS['custom-field-data'], pk
410 @classmethod
411 def _try_lock_data(cls, pk, using):
412 """
413 Take the field's data lock at transaction scope, returning False if it is held elsewhere.
414 Never waits: a job holds this lock for the duration of its bulk update, which may run for
415 hours (see CUSTOMFIELD_JOB_TIMEOUT).
416 """
417 with connections[using].cursor() as cursor:
418 cursor.execute('SELECT pg_try_advisory_xact_lock(%s, %s)', cls.data_lock_key(pk))
419 return cursor.fetchone()[0]
421 def _lock_status(self, using):
422 """
423 Re-read the field's status under a row lock, returning None where the row no longer exists.
425 The status is not taken from this instance, which a job or a concurrent request may have
426 changed since it was fetched, and which must not change between being checked by the caller
427 and the field being marked below.
428 """
429 return self.__class__.objects.using(using).select_for_update().filter(
430 pk=self.pk
431 ).values_list('status', flat=True).first()
433 @staticmethod
434 def _update_object_data(model, filters=None, commit_per_batch=False, **update_kwargs):
435 """
436 Apply an UPDATE to the custom_field_data of every instance of the given model, in batches
437 of at most BULK_UPDATE_CHUNK_SIZE rows. Bounding the number of rows touched by each statement
438 keeps a very large table from exceeding the database statement timeout, as a JSONB update
439 rewrites each affected row in full.
441 :param filters: Optional Q object restricting which rows are updated. Negate it to address
442 the rows which do not match instead.
443 :param commit_per_batch: Commit each batch independently rather than wrapping them all in a
444 single transaction, so that a long-running job does not hold row locks for its whole
445 duration. Only for updates which can safely be resumed.
446 """
447 return chunked_update(
448 model.objects.filter(filters or Q()),
449 commit_per_batch=commit_per_batch,
450 **update_kwargs,
451 )
453 @staticmethod
454 def _exceeds_inline_limit(content_types):
455 """
456 Return True if a bulk update of custom field data across the given object types is too large
457 to perform within the request which triggered it, and must be handed to a background job
458 instead. The limit is BULK_UPDATE_CHUNK_SIZE objects across all of the given types: an
459 update which fits within a single statement is comfortably within any request timeout.
461 The rows are probed rather than counted: `COUNT(*)` reads the whole table, whereas counting
462 one primary key more than the limit costs the same on a table of ten million rows as on one
463 of ten thousand. Only the primary key is selected, and the model's default ordering cleared,
464 to keep the probe to an index-only scan.
466 On the deletion path this over-estimates, as every row of the type is counted where
467 remove_stale_data() would rewrite only those holding the field's key. Probing the key
468 instead would match the work exactly, but custom_field_data carries no index, so the LIMIT
469 could not bound the scan.
470 """
471 # Setting BULK_UPDATE_CHUNK_SIZE to None disables chunking, so the update would be issued
472 # as a single unbounded statement -- precisely what must not run inside a request. Treat any
473 # affected object as exceeding the limit, handing the work to the job, which issues that one
474 # statement under a timeout generous enough to survive it (see CUSTOMFIELD_JOB_TIMEOUT). A
475 # limit of zero leaves the probe below testing for a single row, so a field affecting no
476 # objects still needs no job.
477 limit = settings.BULK_UPDATE_CHUNK_SIZE
478 remaining = 0 if limit is None else limit
480 for ct in content_types:
481 if model := ct.model_class():
482 remaining -= model.objects.order_by().values_list('pk', flat=True)[:remaining + 1].count()
483 if remaining < 0:
484 return True
485 return False
487 def provision_data(self, object_types):
488 """
489 Populate the field's default value across the existing objects of the given object types.
491 Where too many objects are affected to handle within the request, the field is taken offline
492 and the backfill handed to a background job: it does not go live until the job has finished
493 (see CustomFieldStatusChoices).
495 Assignment to a field which is not live is refused, as CustomField.clean() refuses every
496 other change to one: its configuration must not move under the job which is acting on it.
497 Were a second backfill deferred here, it would carry only the object types passed to it, and
498 whichever of the two jobs ran first would bring the field live -- leaving the other to find
499 a field it no longer matched, and its own object types silently unprovisioned.
500 """
501 from extras.jobs import CustomFieldProvisioningJob
503 using = router.db_for_write(self.__class__, instance=self)
505 with transaction.atomic(using=using):
507 # The status is re-read under a row lock rather than taken from this instance
508 self.status = self._lock_status(using)
509 if self.status is None:
510 # Deleted by a concurrent request since this instance was fetched; there is no field
511 # left to assign. Reported rather than ignored, as the assignment has not been applied.
512 raise AbortRequest(
513 _("Custom field '{name}' no longer exists.").format(name=self.name)
514 )
516 if self.status != CustomFieldStatusChoices.STATUS_ACTIVE:
517 raise AbortRequest(
518 _("Custom field '{name}' cannot be assigned to additional object types while its "
519 "stored data is being updated (status: {status}).").format(
520 name=self.name, status=self.get_status_display().lower()
521 )
522 )
524 if self.default is None:
525 return
527 object_types = list(object_types)
528 if not self._exceeds_inline_limit(object_types):
529 self.populate_initial_data(object_types)
530 return
532 self.status = CustomFieldStatusChoices.STATUS_PROVISIONING
533 # Applied via the queryset so that taking the field offline does not itself record a change.
534 self.__class__.objects.using(using).filter(pk=self.pk).update(status=self.status)
535 self.__class__.objects.clear_cache()
537 # Deferred until commit so that the worker cannot observe the field before it is marked.
538 # The types are carried to the job, which cannot otherwise know which of the field's
539 # assignments are the new ones.
540 transaction.on_commit(
541 lambda: CustomFieldProvisioningJob.enqueue_for(
542 self, object_type_pks=[ct.pk for ct in object_types]
543 ),
544 using=using
545 )
547 def remove_data(self, object_types):
548 """
549 Remove the field's stored data from the existing objects of the given object types, as the
550 field is unassigned from them.
552 Unassignment from a field which is not live is refused, as provision_data() refuses an
553 assignment to one. The job acting on the field's data carries the object types it was given
554 and would not observe an unassignment made under it: it would write its defaults into objects
555 the removal had already swept, then bring the field live with values left on objects it no
556 longer applies to.
558 Unlike provisioning and deletion, this is never deferred to a job. Only the objects which
559 actually hold a value for the field are rewritten, which on an unassignment is typically a
560 small fraction of the table (see the note in the custom fields documentation).
561 """
562 using = router.db_for_write(self.__class__, instance=self)
564 with transaction.atomic(using=using):
566 # The status is re-read under a row lock rather than taken from this instance, which a
567 # job may have taken offline since it was fetched, and which must not change between the
568 # check below and the data being removed.
569 self.status = self._lock_status(using)
571 if self.status is None:
572 # Deleted by a concurrent request since this instance was fetched; whatever data
573 # remains belongs to the deletion, which removes it in full.
574 raise AbortRequest(
575 _("Custom field '{name}' no longer exists.").format(name=self.name)
576 )
578 if self.status != CustomFieldStatusChoices.STATUS_ACTIVE:
579 raise AbortRequest(
580 _("Custom field '{name}' cannot be unassigned from object types while its "
581 "stored data is being updated (status: {status}).").format(
582 name=self.name, status=self.get_status_display().lower()
583 )
584 )
586 self.remove_stale_data(object_types)
588 def populate_initial_data(self, content_types, commit_per_batch=False):
589 """
590 Populate initial custom field data upon either a) the creation of a new CustomField, or
591 b) the assignment of an existing CustomField to new object types.
593 Objects which already hold a key for the field are left alone, making this idempotent -- as
594 a retried job requires, and as committing the backfill in batches relies on. (Note that a
595 cleared value is a JSON null rather than an absent key, and so is likewise preserved.)
596 """
597 if self.default is None:
598 return
600 value = Value(self.default, models.JSONField())
601 for ct in content_types:
602 if model := ct.model_class():
603 self._update_object_data(
604 model,
605 filters=~Q(custom_field_data__has_key=self.name),
606 commit_per_batch=commit_per_batch,
607 custom_field_data=Func(
608 F('custom_field_data'),
609 Value([self.name]),
610 value,
611 function='jsonb_set'
612 )
613 )
615 def remove_stale_data(self, content_types, commit_per_batch=False):
616 """
617 Delete custom field data which is no longer relevant (either because the CustomField is
618 no longer assigned to a model, or because it has been deleted).
620 Only objects which actually hold a value for the field are rewritten. That typically excludes
621 the bulk of the table, and makes this idempotent -- as committing the removal in batches
622 relies on -- since a row is dropped from the queryset by the update which removes its key.
623 """
624 for ct in content_types:
625 if model := ct.model_class():
626 self._update_object_data(
627 model,
628 filters=Q(custom_field_data__has_key=self.name),
629 commit_per_batch=commit_per_batch,
630 custom_field_data=F('custom_field_data') - self.name
631 )
633 def rename_object_data(self, old_name, new_name):
634 """
635 Called when a CustomField has been renamed. Removes the original key and inserts the new
636 one, copying the value of the old key.
637 """
638 for ct in self.object_types.all():
639 if model := ct.model_class():
640 self._update_object_data(
641 model,
642 filters=Q(custom_field_data__has_key=old_name),
643 custom_field_data=Func(
644 F('custom_field_data') - old_name,
645 Value([new_name]),
646 Func(
647 F('custom_field_data'),
648 Value(old_name),
649 function='jsonb_extract_path',
650 output_field=models.JSONField()
651 ),
652 function='jsonb_set')
653 )
655 def delete(self, using=None, *args, **kwargs):
656 """
657 Delete the field, deferring the removal of its stored data to a background job where too
658 many objects are affected to handle within the request (see #22996).
660 Where the work is deferred, the row is retained until the job completes: `name` is unique, so
661 for as long as the row exists no other field can take this name and inherit the data still
662 awaiting removal.
664 The deletion signals are dispatched here rather than when the row is finally removed, so that
665 protection rules, the change log, event rules and the search index observe the deletion where
666 the user performed it. They run again in the worker, where every effect beyond the protection
667 rules is gated on there being a current request, making the replay a no-op.
669 The deletion is refused outright if a background job holds the field's data lock, rather than
670 queueing behind that job. This applies equally to a field already pending deletion: reporting
671 a deletion which did not happen would be worse than refusing it. A field stranded in a pending
672 state by a job which never ran holds no lock, and stays deletable; retrying the deletion of
673 one already pending enqueues a fresh purge job for it.
675 Deleting a field already marked for deletion -- by an earlier request of the user's own, or by
676 a concurrent one -- removes nothing further and dispatches no second set of deletion signals.
677 """
678 from extras.jobs import CustomFieldPurgeJob
680 using = using or router.db_for_write(self.__class__, instance=self)
682 with transaction.atomic(using=using):
683 if not self._try_lock_data(self.pk, using):
684 raise AbortRequest(
685 _("Custom field '{name}' is being updated by a background job and cannot be "
686 "deleted until that job has completed.").format(name=self.name)
687 )
689 # The status is re-read under a row lock rather than taken from this instance
690 self.status = self._lock_status(using)
691 if self.status is None:
692 # Already deleted outright by a concurrent request; nothing remains to delete.
693 return 0, {}
695 if self.status == CustomFieldStatusChoices.STATUS_DELETING:
696 # Already pending deletion; the purge job will remove the row once its data is gone.
697 # The lock being free, no job is *running*, so the one enqueued when the field was
698 # marked may never have run: enqueue another, delete() being the only route to one.
699 # Left as it is, a field whose job never ran could never be removed, and would hold
700 # its name against a replacement indefinitely. Where that job is merely queued (a
701 # concurrent deletion having just marked the field), the second job is harmless:
702 # purge_custom_field() rechecks the status under the lock and no-ops.
703 transaction.on_commit(lambda: CustomFieldPurgeJob.enqueue_for(self), using=using)
704 return 0, {}
706 if not self._exceeds_inline_limit(self.object_types.all()):
707 # Few enough objects to purge within the request: delete the row outright, its
708 # stored data being removed by handle_cf_deleted().
709 return super().delete(using, *args, **kwargs)
711 # Update the custom field's status before the signals are dispatched. Applied via the
712 # queryset to avoid emitting a spurious "updated" change record.
713 self.status = CustomFieldStatusChoices.STATUS_DELETING
714 self.__class__.objects.using(using).filter(pk=self.pk).update(status=self.status)
715 self.__class__.objects.clear_cache()
717 models.signals.pre_delete.send(sender=self.__class__, instance=self, using=using, origin=self)
718 models.signals.post_delete.send(sender=self.__class__, instance=self, using=using, origin=self)
720 # Deferred until commit so that the worker cannot observe the field before it is marked,
721 # and is not enqueued at all if the deletion is aborted.
722 transaction.on_commit(lambda: CustomFieldPurgeJob.enqueue_for(self), using=using)
724 return 1, {self._meta.label: 1}
726 def _delete_row(self):
727 """
728 Remove the row itself. Called by CustomFieldPurgeJob once the field's stored data has been
729 purged; nothing else should bypass delete().
730 """
731 return super().delete()
733 def clean(self):
734 super().clean()
736 # A field awaiting a bulk update of its stored data is not live, and its configuration must
737 # not change under the job which is acting on it.
738 if self.pk and self.status != CustomFieldStatusChoices.STATUS_ACTIVE:
739 raise ValidationError(
740 _("Custom field '{name}' cannot be modified while its stored data is being updated "
741 "(status: {status}).").format(name=self.name, status=self.get_status_display().lower())
742 )
744 # Validate the field's default value (if any)
745 if self.default is not None:
746 try:
747 if self.type in (CustomFieldTypeChoices.TYPE_TEXT, CustomFieldTypeChoices.TYPE_LONGTEXT):
748 default_value = str(self.default)
749 else:
750 default_value = self.default
751 self.validate(default_value)
752 except ValidationError as err:
753 raise ValidationError({
754 'default': _(
755 'Invalid default value "{value}": {error}'
756 ).format(value=self.default, error=err.message)
757 })
759 # Minimum/maximum values can be set only for numeric fields
760 if self.type not in (CustomFieldTypeChoices.TYPE_INTEGER, CustomFieldTypeChoices.TYPE_DECIMAL):
761 if self.validation_minimum:
762 raise ValidationError({'validation_minimum': _("A minimum value may be set only for numeric fields")})
763 if self.validation_maximum:
764 raise ValidationError({'validation_maximum': _("A maximum value may be set only for numeric fields")})
766 # Regex validation can be set only for text fields
767 regex_types = (
768 CustomFieldTypeChoices.TYPE_TEXT,
769 CustomFieldTypeChoices.TYPE_LONGTEXT,
770 CustomFieldTypeChoices.TYPE_URL,
771 )
772 if self.validation_regex and self.type not in regex_types:
773 raise ValidationError({
774 'validation_regex': _("Regular expression validation is supported only for text and URL fields")
775 })
777 # Schema validation can be set only for JSON fields
778 if self.validation_schema and self.type != CustomFieldTypeChoices.TYPE_JSON:
779 raise ValidationError({
780 'validation_schema': _("JSON schema validation is supported only for JSON fields")
781 })
783 # Uniqueness can not be enforced for boolean fields
784 if self.unique and self.type == CustomFieldTypeChoices.TYPE_BOOLEAN:
785 raise ValidationError({
786 'unique': _("Uniqueness cannot be enforced for boolean fields")
787 })
789 # Choice set must be set on selection fields, and *only* on selection fields
790 if self.type in (
791 CustomFieldTypeChoices.TYPE_SELECT,
792 CustomFieldTypeChoices.TYPE_MULTISELECT
793 ):
794 if not self.choice_set:
795 raise ValidationError({
796 'choice_set': _("Selection fields must specify a set of choices.")
797 })
798 elif self.choice_set:
799 raise ValidationError({
800 'choice_set': _("Choices may be set only on selection fields.")
801 })
803 # Object fields must define an object_type; other fields must not
804 if self.type in (CustomFieldTypeChoices.TYPE_OBJECT, CustomFieldTypeChoices.TYPE_MULTIOBJECT):
805 if not self.related_object_type:
806 raise ValidationError({
807 'related_object_type': _("Object fields must define an object type.")
808 })
809 elif self.related_object_type:
810 raise ValidationError({
811 'type': _("{type} fields may not define an object type.") .format(type=self.get_type_display())
812 })
814 # Related object filter can be set only for object-type fields, and must contain a dictionary mapping (if set)
815 if self.related_object_filter is not None:
816 if self.type not in (CustomFieldTypeChoices.TYPE_OBJECT, CustomFieldTypeChoices.TYPE_MULTIOBJECT):
817 raise ValidationError({
818 'related_object_filter': _("A related object filter can be defined only for object fields.")
819 })
820 if type(self.related_object_filter) is not dict:
821 raise ValidationError({
822 'related_object_filter': _("Filter must be defined as a dictionary mapping attributes to values.")
823 })
825 def serialize(self, value):
826 """
827 Prepare a value for storage as JSON data.
828 """
829 if value is None:
830 return value
831 if self.type == CustomFieldTypeChoices.TYPE_DECIMAL:
832 return float(value)
833 if self.type == CustomFieldTypeChoices.TYPE_DATE and type(value) is date:
834 return value.isoformat()
835 if self.type == CustomFieldTypeChoices.TYPE_DATETIME and type(value) is datetime:
836 return value.isoformat()
837 if self.type == CustomFieldTypeChoices.TYPE_OBJECT:
838 return value.pk
839 if self.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT:
840 return [obj.pk for obj in value] or None
841 return value
843 def deserialize(self, value):
844 """
845 Convert JSON data to a Python object suitable for the field type.
846 """
847 if value is None:
848 return value
849 if self.type == CustomFieldTypeChoices.TYPE_DATE:
850 try:
851 return date.fromisoformat(value)
852 except ValueError:
853 return value
854 if self.type == CustomFieldTypeChoices.TYPE_DATETIME:
855 try:
856 return datetime.fromisoformat(value)
857 except ValueError:
858 return value
859 if self.type == CustomFieldTypeChoices.TYPE_OBJECT:
860 model = self.related_object_type.model_class()
861 return model.objects.filter(pk=value).first()
862 if self.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT:
863 model = self.related_object_type.model_class()
864 return model.objects.filter(pk__in=value)
865 return value
867 def to_form_field(
868 self,
869 set_initial=True,
870 enforce_required=True,
871 enforce_visibility=True,
872 for_csv_import=False,
873 for_filterset_form=False,
874 ):
875 """
876 Return a form field suitable for setting a CustomField's value for an object.
878 set_initial: Set initial data for the field. This should be False when generating a field for bulk editing.
879 enforce_required: Honor the value of CustomField.required. Set to False for filtering/bulk editing.
880 enforce_visibility: Honor the value of CustomField.ui_visible. Set to False for filtering.
881 for_csv_import: Return a form field suitable for bulk import of objects in CSV format.
882 for_filterset_form: Return a form field suitable for use in a FilterSet form.
883 """
884 initial = self.default if set_initial else None
885 required = self.required if enforce_required else False
887 # Integer
888 if self.type == CustomFieldTypeChoices.TYPE_INTEGER:
889 field = forms.IntegerField(
890 required=required,
891 initial=initial,
892 min_value=self.validation_minimum,
893 max_value=self.validation_maximum
894 )
896 # Decimal
897 elif self.type == CustomFieldTypeChoices.TYPE_DECIMAL:
898 field = forms.DecimalField(
899 required=required,
900 initial=initial,
901 max_digits=16,
902 decimal_places=4,
903 min_value=self.validation_minimum,
904 max_value=self.validation_maximum
905 )
907 # Boolean
908 elif self.type == CustomFieldTypeChoices.TYPE_BOOLEAN:
909 choices = (
910 (None, '---------'),
911 (True, _('True')),
912 (False, _('False')),
913 )
914 field = forms.NullBooleanField(
915 required=required, initial=initial, widget=forms.Select(choices=choices)
916 )
918 # Date
919 elif self.type == CustomFieldTypeChoices.TYPE_DATE:
920 field = forms.DateField(required=required, initial=initial, widget=DatePicker())
922 # Date & time
923 elif self.type == CustomFieldTypeChoices.TYPE_DATETIME:
924 field = forms.DateTimeField(required=required, initial=initial, widget=DateTimePicker())
926 # Select
927 elif self.type in (CustomFieldTypeChoices.TYPE_SELECT, CustomFieldTypeChoices.TYPE_MULTISELECT):
928 choices = self.choice_set.choices
929 default_choice = self.default if self.default in self.choices else None
931 if not required or default_choice is None:
932 choices = add_blank_choice(choices)
934 # Set the initial value to the first available choice (if any)
935 if set_initial and default_choice:
936 initial = default_choice
938 if for_csv_import:
939 if self.type == CustomFieldTypeChoices.TYPE_SELECT:
940 field_class = CSVChoiceField
941 else:
942 field_class = CSVMultipleChoiceField
943 field = field_class(choices=choices, required=required, initial=initial)
944 else:
945 if self.type == CustomFieldTypeChoices.TYPE_SELECT and not for_filterset_form:
946 field_class = DynamicChoiceField
947 widget_class = APISelect
948 else:
949 field_class = DynamicMultipleChoiceField
950 widget_class = APISelectMultiple
951 field = field_class(
952 choices=choices,
953 required=required,
954 initial=initial,
955 widget=widget_class(api_url=f'/api/extras/custom-field-choice-sets/{self.choice_set.pk}/choices/')
956 )
958 # URL
959 elif self.type == CustomFieldTypeChoices.TYPE_URL:
960 field = LaxURLField(assume_scheme='https', required=required, initial=initial)
961 if self.validation_regex:
962 field.validators = [
963 RegexValidator(
964 regex=self.validation_regex,
965 message=mark_safe(_("Values must match this regex: <code>{regex}</code>").format(
966 regex=escape(self.validation_regex)
967 ))
968 )
969 ]
971 # JSON
972 elif self.type == CustomFieldTypeChoices.TYPE_JSON:
973 field = JSONField(required=required, initial=json.dumps(initial) if initial is not None else None)
975 # Object
976 elif self.type == CustomFieldTypeChoices.TYPE_OBJECT:
977 model = self.related_object_type.model_class()
978 if for_csv_import:
979 field_class = CSVModelChoiceField
980 elif for_filterset_form:
981 field_class = DynamicModelMultipleChoiceField
982 else:
983 field_class = DynamicModelChoiceField
984 kwargs = {
985 'queryset': model.objects.all(),
986 'required': required,
987 'initial': initial,
988 }
989 if not for_csv_import:
990 kwargs['query_params'] = self.related_object_filter
991 kwargs['selector'] = True
993 field = field_class(**kwargs)
995 # Multiple objects
996 elif self.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT:
997 model = self.related_object_type.model_class()
998 field_class = CSVModelMultipleChoiceField if for_csv_import else DynamicModelMultipleChoiceField
999 kwargs = {
1000 'queryset': model.objects.all(),
1001 'required': required,
1002 'initial': initial,
1003 }
1004 if not for_csv_import:
1005 kwargs['query_params'] = self.related_object_filter
1006 kwargs['selector'] = True
1008 field = field_class(**kwargs)
1010 # Text
1011 else:
1012 widget = forms.Textarea if self.type == CustomFieldTypeChoices.TYPE_LONGTEXT else None
1013 field = forms.CharField(required=required, initial=initial, widget=widget)
1014 if self.validation_regex:
1015 field.validators = [
1016 RegexValidator(
1017 regex=self.validation_regex,
1018 message=mark_safe(_("Values must match this regex: <code>{regex}</code>").format(
1019 regex=escape(self.validation_regex)
1020 ))
1021 )
1022 ]
1024 field.model = self
1025 field.label = str(self)
1026 if self.description:
1027 field.help_text = render_markdown(self.description)
1029 # Annotate read-only fields
1030 if enforce_visibility and self.ui_editable != CustomFieldUIEditableChoices.YES:
1031 field.disabled = True
1033 return field
1035 def to_filter(self, lookup_expr=None):
1036 """
1037 Return a django_filters Filter instance suitable for this field type.
1039 :param lookup_expr: Custom lookup expression (optional)
1040 """
1041 # Imported locally as extras.filters imports extras.models
1042 from extras.filters import missing_key_aware_filter_factory
1044 kwargs = {
1045 'field_name': f'custom_field_data__{self.name}'
1046 }
1047 # Native numeric filters will use `isnull` by default for empty lookups, but
1048 # JSON fields require `empty` (see bug #20012).
1049 if lookup_expr == 'isnull':
1050 lookup_expr = 'empty'
1051 if lookup_expr is not None:
1052 kwargs['lookup_expr'] = lookup_expr
1054 # 'Empty' lookup is always a boolean
1055 if lookup_expr == 'empty':
1056 filter_class = django_filters.BooleanFilter
1058 # Text/URL
1059 elif self.type in (
1060 CustomFieldTypeChoices.TYPE_TEXT,
1061 CustomFieldTypeChoices.TYPE_LONGTEXT,
1062 CustomFieldTypeChoices.TYPE_URL,
1063 ):
1064 filter_class = filters.MultiValueCharFilter
1065 if self.filter_logic == CustomFieldFilterLogicChoices.FILTER_LOOSE:
1066 kwargs['lookup_expr'] = 'icontains'
1068 # Integer
1069 elif self.type == CustomFieldTypeChoices.TYPE_INTEGER:
1070 filter_class = filters.MultiValueNumberFilter
1072 # Decimal
1073 elif self.type == CustomFieldTypeChoices.TYPE_DECIMAL:
1074 filter_class = filters.MultiValueDecimalFilter
1076 # Boolean
1077 elif self.type == CustomFieldTypeChoices.TYPE_BOOLEAN:
1078 filter_class = django_filters.BooleanFilter
1080 # Date
1081 elif self.type == CustomFieldTypeChoices.TYPE_DATE:
1082 filter_class = filters.MultiValueDateFilter
1084 # Date & time
1085 elif self.type == CustomFieldTypeChoices.TYPE_DATETIME:
1086 filter_class = filters.MultiValueDateTimeFilter
1088 # Select
1089 elif self.type == CustomFieldTypeChoices.TYPE_SELECT:
1090 filter_class = filters.MultiValueCharFilter
1092 # Multiselect
1093 elif self.type == CustomFieldTypeChoices.TYPE_MULTISELECT:
1094 # Do not pin lookup_expr: FILTER_ARRAY_BASED_LOOKUP_MAP preserves the class default under negation
1095 filter_class = filters.MultiValueArrayFilter
1097 # Object
1098 elif self.type == CustomFieldTypeChoices.TYPE_OBJECT:
1099 filter_class = filters.MultiValueNumberFilter
1101 # Multi-object
1102 elif self.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT:
1103 filter_class = filters.MultiValueNumberFilter
1104 kwargs['lookup_expr'] = 'contains'
1106 # Unsupported custom field type
1107 else:
1108 return None
1110 # A negated lookup must match objects which carry no key for this field at all; see
1111 # MissingKeyAwareFilterMixin. BooleanFilter is never negated, so it is left alone.
1112 if not issubclass(filter_class, django_filters.BooleanFilter):
1113 filter_class = missing_key_aware_filter_factory(filter_class)
1115 filter_instance = filter_class(**kwargs)
1116 filter_instance.custom_field = self
1118 return filter_instance
1120 def validate(self, value):
1121 """
1122 Validate a value according to the field's type validation rules.
1123 """
1124 if value not in [None, '']:
1126 # Validate text field
1127 if self.type in (CustomFieldTypeChoices.TYPE_TEXT, CustomFieldTypeChoices.TYPE_LONGTEXT):
1128 if type(value) is not str:
1129 raise ValidationError(_("Value must be a string."))
1130 if self.validation_regex and not re.match(self.validation_regex, value):
1131 raise ValidationError(_("Value must match regex '{regex}'").format(regex=self.validation_regex))
1133 # Validate URL field
1134 elif self.type == CustomFieldTypeChoices.TYPE_URL:
1135 if type(value) is not str:
1136 raise ValidationError(_("Value must be a string."))
1137 # Enforce ALLOWED_URL_SCHEMES to guard against dangerous schemes (e.g. javascript:). A
1138 # schemeless value is permitted and treated as relative.
1139 if not url_scheme_is_allowed(value):
1140 raise ValidationError(
1141 _("URLs must use a scheme permitted by ALLOWED_URL_SCHEMES.")
1142 )
1143 if self.validation_regex and not re.match(self.validation_regex, value):
1144 raise ValidationError(_("Value must match regex '{regex}'").format(regex=self.validation_regex))
1146 # Validate integer
1147 elif self.type == CustomFieldTypeChoices.TYPE_INTEGER:
1148 if type(value) is not int:
1149 raise ValidationError(_("Value must be an integer."))
1150 if self.validation_minimum is not None and value < self.validation_minimum:
1151 raise ValidationError(
1152 _("Value must be at least {minimum}").format(minimum=self.validation_minimum)
1153 )
1154 if self.validation_maximum is not None and value > self.validation_maximum:
1155 raise ValidationError(
1156 _("Value must not exceed {maximum}").format(maximum=self.validation_maximum)
1157 )
1159 # Validate decimal
1160 elif self.type == CustomFieldTypeChoices.TYPE_DECIMAL:
1161 try:
1162 decimal.Decimal(value)
1163 except decimal.InvalidOperation:
1164 raise ValidationError(_("Value must be a decimal."))
1165 if self.validation_minimum is not None and value < self.validation_minimum:
1166 raise ValidationError(
1167 _("Value must be at least {minimum}").format(minimum=self.validation_minimum)
1168 )
1169 if self.validation_maximum is not None and value > self.validation_maximum:
1170 raise ValidationError(
1171 _("Value must not exceed {maximum}").format(maximum=self.validation_maximum)
1172 )
1174 # Validate boolean
1175 elif self.type == CustomFieldTypeChoices.TYPE_BOOLEAN and value not in [True, False, 1, 0]:
1176 raise ValidationError(_("Value must be true or false."))
1178 # Validate date
1179 elif self.type == CustomFieldTypeChoices.TYPE_DATE:
1180 if type(value) is not date:
1181 try:
1182 date.fromisoformat(value)
1183 except ValueError:
1184 raise ValidationError(_("Date values must be in ISO 8601 format (YYYY-MM-DD)."))
1186 # Validate date & time
1187 elif self.type == CustomFieldTypeChoices.TYPE_DATETIME:
1188 if type(value) is not datetime:
1189 try:
1190 datetime_from_timestamp(value)
1191 except ValueError:
1192 raise ValidationError(
1193 _("Date and time values must be in ISO 8601 format (YYYY-MM-DD HH:MM:SS).")
1194 )
1196 # Validate selected choice
1197 elif self.type == CustomFieldTypeChoices.TYPE_SELECT:
1198 if value not in self.choice_set.values:
1199 raise ValidationError(
1200 _("Invalid choice ({value}) for choice set {choiceset}.").format(
1201 value=value,
1202 choiceset=self.choice_set
1203 )
1204 )
1206 # Validate all selected choices
1207 elif self.type == CustomFieldTypeChoices.TYPE_MULTISELECT:
1208 # Require a list of valid string choices. The isinstance() check short-circuits the membership
1209 # test so that non-string members (e.g. a client echoing back the {value, label} read
1210 # representation) raise a ValidationError rather than an unhashable-type TypeError.
1211 valid_values = set(self.choice_set.values)
1212 if type(value) is not list or not all(isinstance(v, str) and v in valid_values for v in value):
1213 raise ValidationError(
1214 _("Invalid choice(s) ({value}) for choice set {choiceset}.").format(
1215 value=value,
1216 choiceset=self.choice_set
1217 )
1218 )
1220 # Validate selected object
1221 elif self.type == CustomFieldTypeChoices.TYPE_OBJECT:
1222 if type(value) is not int:
1223 raise ValidationError(_("Value must be an object ID, not {type}").format(type=type(value).__name__))
1225 # Validate selected objects
1226 elif self.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT:
1227 if type(value) is not list:
1228 raise ValidationError(
1229 _("Value must be a list of object IDs, not {type}").format(type=type(value).__name__)
1230 )
1231 for id in value:
1232 if type(id) is not int:
1233 raise ValidationError(_("Found invalid object ID: {id}").format(id=id))
1235 # Validate JSON against schema (if defined)
1236 elif self.type == CustomFieldTypeChoices.TYPE_JSON:
1237 if self.validation_schema:
1238 try:
1239 jsonschema.validate(value, schema=self.validation_schema)
1240 except JSONValidationError as e:
1241 raise ValidationError(
1242 _("Value does not conform to the assigned schema: {error}").format(error=e.message)
1243 )
1245 elif self.required:
1246 raise ValidationError(_("Required field cannot be empty."))
1249class CustomFieldChoiceSet(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedModel):
1250 """
1251 Represents a set of choices available for choice and multi-choice custom fields.
1252 """
1253 name = models.CharField(
1254 max_length=100,
1255 unique=True
1256 )
1257 description = models.CharField(
1258 max_length=200,
1259 blank=True
1260 )
1261 base_choices = models.CharField(
1262 max_length=50,
1263 choices=CustomFieldChoiceSetBaseChoices,
1264 blank=True,
1265 null=True,
1266 help_text=_('Base set of predefined choices (optional)')
1267 )
1268 extra_choices = ChoiceSetField(
1269 blank=True,
1270 null=True
1271 )
1272 choice_colors = models.JSONField(
1273 default=dict,
1274 blank=True,
1275 )
1276 order_alphabetically = models.BooleanField(
1277 default=False,
1278 help_text=_('Choices are automatically ordered alphabetically')
1279 )
1281 clone_fields = ('extra_choices', 'choice_colors', 'order_alphabetically')
1283 class Meta:
1284 ordering = ('name',)
1285 verbose_name = _('custom field choice set')
1286 verbose_name_plural = _('custom field choice sets')
1288 def __str__(self):
1289 return self.name
1291 def __init__(self, *args, **kwargs):
1292 super().__init__(*args, **kwargs)
1294 # Cache the initial set of choices for comparison under clean()
1295 self._original_extra_choices = self.__dict__.get('extra_choices')
1297 def get_absolute_url(self):
1298 return reverse('extras:customfieldchoiceset', args=[self.pk])
1300 @property
1301 def choices(self):
1302 """
1303 Returns a concatenation of the base and extra choices.
1304 """
1305 if not hasattr(self, '_choices'):
1306 self._choices = []
1307 if self.base_choices:
1308 self._choices.extend(CHOICE_SETS.get(self.base_choices))
1309 if self.extra_choices:
1310 self._choices.extend(self.extra_choices)
1311 if self.order_alphabetically:
1312 self._choices = sorted(self._choices, key=lambda x: x[0])
1313 return self._choices
1315 @property
1316 def colors(self):
1317 """
1318 Return merged color mappings from the selected base choice set (if it defines colors)
1319 and any custom color overrides defined on this choice set.
1320 """
1321 if not hasattr(self, '_colors'):
1322 self._colors = {}
1323 if self.base_choices:
1324 base_choice_set = CHOICE_SETS.get(self.base_choices)
1325 self._colors.update(getattr(base_choice_set, 'colors', {}))
1326 if self.choice_colors:
1327 self._colors.update(self.choice_colors)
1328 return self._colors
1330 def get_choice_color(self, value):
1331 return self.colors.get(value)
1333 @property
1334 def choices_count(self):
1335 return len(self.choices)
1337 @property
1338 def values(self):
1339 """
1340 Returns an iterator of the valid choice values.
1341 """
1342 return (x[0] for x in self.choices)
1344 def clean(self):
1345 if not self.base_choices and not self.extra_choices:
1346 raise ValidationError(_("Must define base or extra choices."))
1348 if self.choice_colors is None:
1349 self.choice_colors = {}
1350 elif not isinstance(self.choice_colors, dict):
1351 raise ValidationError({
1352 'choice_colors': _('Color mappings must be defined as a JSON object.')
1353 })
1355 valid_choice_values = set()
1356 extra_choice_values = set()
1358 if self.base_choices:
1359 valid_choice_values.update(value for value, _ in CHOICE_SETS.get(self.base_choices))
1361 if self.extra_choices:
1362 for value, _label in self.extra_choices:
1363 if value in extra_choice_values:
1364 raise ValidationError(_("Duplicate value '{value}' found in extra choices.").format(value=value))
1365 extra_choice_values.add(value)
1366 valid_choice_values.update(extra_choice_values)
1368 invalid_choice_values = set()
1369 invalid_colors = set()
1370 valid_colors = set(CustomFieldChoiceColorChoices.values())
1372 for value, color in self.choice_colors.items():
1373 if value not in valid_choice_values:
1374 invalid_choice_values.add(value)
1375 if color not in valid_colors:
1376 invalid_colors.add(color)
1378 if invalid_choice_values:
1379 raise ValidationError({
1380 'choice_colors': _(
1381 'Color mappings must reference an existing choice value. Invalid value(s): {values}.'
1382 ).format(values=', '.join(sorted(invalid_choice_values)))
1383 })
1385 if invalid_colors:
1386 raise ValidationError({
1387 'choice_colors': _(
1388 'Invalid color value(s): {colors}. Use a supported named color.'
1389 ).format(colors=', '.join(sorted(invalid_colors)))
1390 })
1392 # Check whether any choices have been removed. If so, check whether any of the removed
1393 # choices are still set in custom field data for any object.
1394 original_choices = set([
1395 c[0] for c in self._original_extra_choices
1396 ]) if self._original_extra_choices else set()
1397 if removed_choices := original_choices - valid_choice_values:
1398 for custom_field in self.choices_for.all():
1399 for object_type in custom_field.object_types.all():
1400 model = object_type.model_class()
1401 for choice in removed_choices:
1402 # Form the query based on the type of custom field
1403 if custom_field.type == CustomFieldTypeChoices.TYPE_MULTISELECT:
1404 query_args = {f"custom_field_data__{custom_field.name}__contains": choice}
1405 else:
1406 query_args = {f"custom_field_data__{custom_field.name}": choice}
1407 # Raise a ValidationError if there are any objects which still reference the removed choice
1408 if model.objects.filter(models.Q(**query_args)).exists():
1409 raise ValidationError(
1410 _(
1411 "Cannot remove choice {choice} as there are {model} objects which reference it."
1412 ).format(choice=choice, model=object_type)
1413 )
1415 def save(self, *args, **kwargs):
1417 # Sort choices if alphabetical ordering is enforced
1418 if self.order_alphabetically and self.extra_choices:
1419 self.extra_choices = sorted(self.extra_choices, key=lambda x: x[0])
1421 return super().save(*args, **kwargs)