1from __future__ import annotations
2
3import collections.abc
4import copy
5import datetime
6import decimal
7import enum
8import operator
9import uuid
10import warnings
11from base64 import b64decode, b64encode
12from collections.abc import Callable, Sequence
13from functools import cached_property, total_ordering
14from typing import TYPE_CHECKING, Any
15
16from plain import exceptions, validators
17from plain.models.constants import LOOKUP_SEP
18from plain.models.db import db_connection
19from plain.models.enums import ChoicesMeta
20from plain.models.query_utils import DeferredAttribute, RegisterLookupMixin
21from plain.preflight import PreflightResult
22from plain.utils import timezone
23from plain.utils.datastructures import DictWrapper
24from plain.utils.dateparse import (
25 parse_date,
26 parse_datetime,
27 parse_duration,
28 parse_time,
29)
30from plain.utils.duration import duration_microseconds, duration_string
31from plain.utils.functional import Promise
32from plain.utils.ipv6 import clean_ipv6_address
33from plain.utils.itercompat import is_iterable
34
35from ..registry import models_registry
36
37if TYPE_CHECKING:
38 from plain.models.backends.base.base import BaseDatabaseWrapper
39 from plain.models.fields.reverse_related import ForeignObjectRel
40 from plain.models.sql.compiler import SQLCompiler
41
42__all__ = [
43 "BLANK_CHOICE_DASH",
44 "PrimaryKeyField",
45 "BigIntegerField",
46 "BinaryField",
47 "BooleanField",
48 "CharField",
49 "DateField",
50 "DateTimeField",
51 "DecimalField",
52 "DurationField",
53 "EmailField",
54 "Empty",
55 "Field",
56 "FloatField",
57 "GenericIPAddressField",
58 "IntegerField",
59 "NOT_PROVIDED",
60 "PositiveBigIntegerField",
61 "PositiveIntegerField",
62 "PositiveSmallIntegerField",
63 "SmallIntegerField",
64 "TextField",
65 "TimeField",
66 "URLField",
67 "UUIDField",
68]
69
70
71class Empty:
72 pass
73
74
75class NOT_PROVIDED:
76 pass
77
78
79# The values to use for "blank" in SelectFields. Will be appended to the start
80# of most "choices" lists.
81BLANK_CHOICE_DASH = [("", "---------")]
82
83
84def _load_field(package_label: str, model_name: str, field_name: str) -> Field:
85 return models_registry.get_model(package_label, model_name)._model_meta.get_field(
86 field_name
87 )
88
89
90# A guide to Field parameters:
91#
92# * name: The name of the field specified in the model.
93# * attname: The attribute to use on the model object. This is the same as
94# "name", except in the case of ForeignKeys, where "_id" is
95# appended.
96# * db_column: The db_column specified in the model (or None).
97# * column: The database column for this field. This is the same as
98# "attname", except if db_column is specified.
99#
100# Code that introspects values, or does other dynamic things, should use
101# attname.
102
103
104def _empty(of_cls: type) -> Empty:
105 new = Empty()
106 new.__class__ = of_cls
107 return new
108
109
110def return_None() -> None:
111 return None
112
113
114@total_ordering
115class Field(RegisterLookupMixin):
116 """Base class for all field types"""
117
118 # Designates whether empty strings fundamentally are allowed at the
119 # database level.
120 empty_strings_allowed = True
121 empty_values = list(validators.EMPTY_VALUES)
122
123 # These track each time a Field instance is created. Used to retain order.
124 # The auto_creation_counter is used for fields that Plain implicitly
125 # creates, creation_counter is used for all user-specified fields.
126 creation_counter = 0
127 auto_creation_counter = -1
128 default_validators = [] # Default set of validators
129 default_error_messages = {
130 "invalid_choice": "Value %(value)r is not a valid choice.",
131 "allow_null": "This field cannot be null.",
132 "required": "This field is be required.",
133 "unique": "A %(model_name)s with this %(field_label)s already exists.",
134 }
135
136 # Attributes that don't affect a column definition.
137 # These attributes are ignored when altering the field.
138 non_db_attrs = (
139 "required",
140 "choices",
141 "db_column",
142 "error_messages",
143 "limit_choices_to",
144 # Database-level options are not supported, see #21961.
145 "on_delete",
146 "related_name",
147 "related_query_name",
148 "validators",
149 )
150
151 # Field flags
152 hidden = False
153
154 many_to_many = None
155 many_to_one = None
156 one_to_many = None
157 related_model = None
158
159 descriptor_class = DeferredAttribute
160
161 # Generic field type description, usually overridden by subclasses
162 def _description(self) -> str:
163 return f"Field of type: {self.__class__.__name__}"
164
165 description = property(_description)
166
167 def __init__(
168 self,
169 *,
170 max_length: int | None = None,
171 required: bool = True,
172 allow_null: bool = False,
173 rel: ForeignObjectRel | None = None,
174 default: Any = NOT_PROVIDED,
175 choices: Any = None,
176 db_column: str | None = None,
177 validators: Sequence[Callable[..., Any]] = (),
178 error_messages: dict[str, str] | None = None,
179 db_comment: str | None = None,
180 ):
181 self.name = None # Set by set_attributes_from_name
182 self.max_length = max_length
183 self.required, self.allow_null = required, allow_null
184 self.remote_field = rel
185 self.is_relation = self.remote_field is not None
186 self.default = default
187 if isinstance(choices, ChoicesMeta):
188 choices = choices.choices
189 elif isinstance(choices, enum.EnumMeta):
190 choices = [(member.value, member.name) for member in choices]
191 if isinstance(choices, collections.abc.Iterator):
192 choices = list(choices)
193 self.choices = choices
194 self.db_column = db_column
195 self.db_comment = db_comment
196
197 self.primary_key = False
198 self.auto_created = False
199
200 # Adjust the appropriate creation counter, and save our local copy.
201 self.creation_counter = Field.creation_counter
202 Field.creation_counter += 1
203
204 self._validators = list(validators) # Store for deconstruction later
205
206 self._error_messages = error_messages # Store for deconstruction later
207
208 def __str__(self) -> str:
209 """
210 Return "package_label.model_label.field_name" for fields attached to
211 models.
212 """
213 if not hasattr(self, "model"):
214 return super().__str__()
215 model = self.model
216 return f"{model.model_options.label}.{self.name}"
217
218 def __repr__(self) -> str:
219 """Display the module, class, and name of the field."""
220 path = f"{self.__class__.__module__}.{self.__class__.__qualname__}"
221 name = getattr(self, "name", None)
222 if name is not None:
223 return f"<{path}: {name}>"
224 return f"<{path}>"
225
226 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
227 return [
228 *self._check_field_name(),
229 *self._check_choices(),
230 *self._check_db_comment(),
231 *self._check_null_allowed_for_primary_keys(),
232 *self._check_backend_specific_checks(),
233 *self._check_validators(),
234 ]
235
236 def _check_field_name(self) -> list[PreflightResult]:
237 """
238 Check if field name is valid, i.e. 1) does not end with an
239 underscore, 2) does not contain "__" and 3) is not "id".
240 """
241 if self.name.endswith("_"):
242 return [
243 PreflightResult(
244 fix="Field names must not end with an underscore.",
245 obj=self,
246 id="fields.name_ends_with_underscore",
247 )
248 ]
249 elif LOOKUP_SEP in self.name:
250 return [
251 PreflightResult(
252 fix=f'Field names must not contain "{LOOKUP_SEP}".',
253 obj=self,
254 id="fields.name_contains_lookup_separator",
255 )
256 ]
257 elif self.name == "id":
258 return [
259 PreflightResult(
260 fix="'id' is a reserved word that cannot be used as a field name.",
261 obj=self,
262 id="fields.reserved_field_name_id",
263 )
264 ]
265 else:
266 return []
267
268 @classmethod
269 def _choices_is_value(cls, value: Any) -> bool:
270 return isinstance(value, str | Promise) or not is_iterable(value)
271
272 def _check_choices(self) -> list[PreflightResult]:
273 if not self.choices:
274 return []
275
276 if not is_iterable(self.choices) or isinstance(self.choices, str):
277 return [
278 PreflightResult(
279 fix="'choices' must be an iterable (e.g., a list or tuple).",
280 obj=self,
281 id="fields.choices_not_iterable",
282 )
283 ]
284
285 choice_max_length = 0
286 # Expect [group_name, [value, display]]
287 for choices_group in self.choices:
288 try:
289 group_name, group_choices = choices_group
290 except (TypeError, ValueError):
291 # Containing non-pairs
292 break
293 try:
294 if not all(
295 self._choices_is_value(value) and self._choices_is_value(human_name)
296 for value, human_name in group_choices
297 ):
298 break
299 if self.max_length is not None and group_choices:
300 choice_max_length = max(
301 [
302 choice_max_length,
303 *(
304 len(value)
305 for value, _ in group_choices
306 if isinstance(value, str)
307 ),
308 ]
309 )
310 except (TypeError, ValueError):
311 # No groups, choices in the form [value, display]
312 value, human_name = group_name, group_choices
313 if not self._choices_is_value(value) or not self._choices_is_value(
314 human_name
315 ):
316 break
317 if self.max_length is not None and isinstance(value, str):
318 choice_max_length = max(choice_max_length, len(value))
319
320 # Special case: choices=['ab']
321 if isinstance(choices_group, str):
322 break
323 else:
324 if self.max_length is not None and choice_max_length > self.max_length:
325 return [
326 PreflightResult(
327 fix="'max_length' is too small to fit the longest value " # noqa: UP031
328 "in 'choices' (%d characters)." % choice_max_length,
329 obj=self,
330 id="fields.max_length_too_small_for_choices",
331 ),
332 ]
333 return []
334
335 return [
336 PreflightResult(
337 fix="'choices' must be an iterable containing "
338 "(actual value, human readable name) tuples.",
339 obj=self,
340 id="fields.choices_invalid_format",
341 )
342 ]
343
344 def _check_db_comment(self) -> list[PreflightResult]:
345 if not self.db_comment:
346 return []
347 errors = []
348 if not (
349 db_connection.features.supports_comments
350 or "supports_comments" in self.model.model_options.required_db_features
351 ):
352 errors.append(
353 PreflightResult(
354 fix=f"{db_connection.display_name} does not support comments on "
355 f"columns (db_comment).",
356 obj=self,
357 id="fields.db_comment_unsupported",
358 warning=True,
359 )
360 )
361 return errors
362
363 def _check_null_allowed_for_primary_keys(self) -> list[PreflightResult]:
364 if self.primary_key and self.allow_null:
365 # We cannot reliably check this for backends like Oracle which
366 # consider NULL and '' to be equal (and thus set up
367 # character-based fields a little differently).
368 return [
369 PreflightResult(
370 fix="Primary keys must not have allow_null=True. "
371 "Set allow_null=False on the field, or "
372 "remove primary_key=True argument.",
373 obj=self,
374 id="fields.primary_key_allows_null",
375 )
376 ]
377 else:
378 return []
379
380 def _check_backend_specific_checks(self) -> list[PreflightResult]:
381 errors = []
382 errors.extend(db_connection.validation.check_field(self))
383 return errors
384
385 def _check_validators(self) -> list[PreflightResult]:
386 errors = []
387 for i, validator in enumerate(self.validators):
388 if not callable(validator):
389 errors.append(
390 PreflightResult(
391 fix=(
392 "All 'validators' must be callable. "
393 f"validators[{i}] ({repr(validator)}) isn't a function or "
394 "instance of a validator class."
395 ),
396 obj=self,
397 id="fields.invalid_validator",
398 )
399 )
400 return errors
401
402 def get_col(self, alias: str, output_field: Field | None = None) -> Any:
403 if alias == self.model.model_options.db_table and (
404 output_field is None or output_field == self
405 ):
406 return self.cached_col
407 from plain.models.expressions import Col
408
409 return Col(alias, self, output_field)
410
411 @cached_property
412 def cached_col(self) -> Any:
413 from plain.models.expressions import Col
414
415 return Col(self.model.model_options.db_table, self)
416
417 def select_format(
418 self, compiler: SQLCompiler, sql: str, params: Any
419 ) -> tuple[str, Any]:
420 """
421 Custom format for select clauses. For example, GIS columns need to be
422 selected as AsText(table.col) on MySQL as the table.col data can't be
423 used by Plain.
424 """
425 return sql, params
426
427 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
428 """
429 Return enough information to recreate the field as a 4-tuple:
430
431 * The name of the field on the model, if contribute_to_class() has
432 been run.
433 * The import path of the field, including the class, e.g.
434 plain.models.IntegerField. This should be the most portable
435 version, so less specific may be better.
436 * A list of positional arguments.
437 * A dict of keyword arguments.
438
439 Note that the positional or keyword arguments must contain values of
440 the following types (including inner values of collection types):
441
442 * None, bool, str, int, float, complex, set, frozenset, list, tuple,
443 dict
444 * UUID
445 * datetime.datetime (naive), datetime.date
446 * top-level classes, top-level functions - will be referenced by their
447 full import path
448 * Storage instances - these have their own deconstruct() method
449
450 This is because the values here must be serialized into a text format
451 (possibly new Python code, possibly JSON) and these are the only types
452 with encoding handlers defined.
453
454 There's no need to return the exact way the field was instantiated this
455 time, just ensure that the resulting field is the same - prefer keyword
456 arguments over positional ones, and omit parameters with their default
457 values.
458 """
459 # Short-form way of fetching all the default parameters
460 keywords = {}
461 possibles = {
462 "max_length": None,
463 "required": True,
464 "allow_null": False,
465 "default": NOT_PROVIDED,
466 "choices": None,
467 "db_column": None,
468 "db_comment": None,
469 "validators": [],
470 "error_messages": None,
471 }
472 attr_overrides = {
473 "error_messages": "_error_messages",
474 "validators": "_validators",
475 }
476 equals_comparison = {"choices", "validators"}
477 for name, default in possibles.items():
478 value = getattr(self, attr_overrides.get(name, name))
479 # Unroll anything iterable for choices into a concrete list
480 if name == "choices" and isinstance(value, collections.abc.Iterable):
481 value = list(value)
482 # Do correct kind of comparison
483 if name in equals_comparison:
484 if value != default:
485 keywords[name] = value
486 else:
487 if value is not default:
488 keywords[name] = value
489 # Work out path - we shorten it for known Plain core fields
490 path = f"{self.__class__.__module__}.{self.__class__.__qualname__}"
491 if path.startswith("plain.models.fields.related"):
492 path = path.replace("plain.models.fields.related", "plain.models")
493 elif path.startswith("plain.models.fields.json"):
494 path = path.replace("plain.models.fields.json", "plain.models")
495 elif path.startswith("plain.models.fields.proxy"):
496 path = path.replace("plain.models.fields.proxy", "plain.models")
497 elif path.startswith("plain.models.fields"):
498 path = path.replace("plain.models.fields", "plain.models")
499 # Return basic info - other fields should override this.
500 return (self.name, path, [], keywords)
501
502 def clone(self) -> Field:
503 """
504 Uses deconstruct() to clone a new copy of this Field.
505 Will not preserve any class attachments/attribute names.
506 """
507 name, path, args, kwargs = self.deconstruct()
508 return self.__class__(*args, **kwargs)
509
510 def __eq__(self, other: object) -> bool:
511 # Needed for @total_ordering
512 if isinstance(other, Field):
513 return self.creation_counter == other.creation_counter and getattr(
514 self, "model", None
515 ) == getattr(other, "model", None)
516 return NotImplemented
517
518 def __lt__(self, other: object) -> bool:
519 # This is needed because bisect does not take a comparison function.
520 # Order by creation_counter first for backward compatibility.
521 if isinstance(other, Field):
522 if (
523 self.creation_counter != other.creation_counter
524 or not hasattr(self, "model")
525 and not hasattr(other, "model")
526 ):
527 return self.creation_counter < other.creation_counter
528 elif hasattr(self, "model") != hasattr(other, "model"):
529 return not hasattr(self, "model") # Order no-model fields first
530 else:
531 # creation_counter's are equal, compare only models.
532 return (
533 self.model.model_options.package_label,
534 self.model.model_options.model_name,
535 ) < (
536 other.model.model_options.package_label,
537 other.model.model_options.model_name,
538 )
539 return NotImplemented
540
541 def __hash__(self) -> int:
542 return hash(self.creation_counter)
543
544 def __deepcopy__(self, memodict: dict[int, Any]) -> Field:
545 # We don't have to deepcopy very much here, since most things are not
546 # intended to be altered after initial creation.
547 obj = copy.copy(self)
548 if self.remote_field:
549 obj.remote_field = copy.copy(self.remote_field)
550 if hasattr(self.remote_field, "field") and self.remote_field.field is self:
551 obj.remote_field.field = obj
552 memodict[id(self)] = obj
553 return obj
554
555 def __copy__(self) -> Field:
556 # We need to avoid hitting __reduce__, so define this
557 # slightly weird copy construct.
558 obj = Empty()
559 obj.__class__ = self.__class__
560 obj.__dict__ = self.__dict__.copy()
561 return obj # type: ignore[return-value]
562
563 def __reduce__(
564 self,
565 ) -> (
566 tuple[Callable[..., Any], tuple[Any, ...], dict[str, Any]]
567 | tuple[Callable[..., Field], tuple[str, str, str]]
568 ):
569 """
570 Pickling should return the model._model_meta.fields instance of the field,
571 not a new copy of that field. So, use the app registry to load the
572 model and then the field back.
573 """
574 if not hasattr(self, "model"):
575 # Fields are sometimes used without attaching them to models (for
576 # example in aggregation). In this case give back a plain field
577 # instance. The code below will create a new empty instance of
578 # class self.__class__, then update its dict with self.__dict__
579 # values - so, this is very close to normal pickle.
580 state = self.__dict__.copy()
581 # The _get_default cached_property can't be pickled due to lambda
582 # usage.
583 state.pop("_get_default", None)
584 return _empty, (self.__class__,), state
585 return _load_field, (
586 self.model.model_options.package_label,
587 self.model.model_options.object_name,
588 self.name,
589 )
590
591 def get_id_value_on_save(self, instance: Any) -> Any:
592 """
593 Hook to generate new primary key values on save. This method is called when
594 saving instances with no primary key value set. If this method returns
595 something else than None, then the returned value is used when saving
596 the new instance.
597 """
598 if self.default:
599 return self.get_default()
600 return None
601
602 def to_python(self, value: Any) -> Any:
603 """
604 Convert the input value into the expected Python data type, raising
605 plain.exceptions.ValidationError if the data can't be converted.
606 Return the converted value. Subclasses should override this.
607 """
608 return value
609
610 @cached_property
611 def error_messages(self) -> dict[str, str]:
612 messages = {}
613 for c in reversed(self.__class__.__mro__):
614 messages.update(getattr(c, "default_error_messages", {}))
615 messages.update(self._error_messages or {})
616 return messages
617
618 @cached_property
619 def validators(self) -> list[Callable[..., Any]]:
620 """
621 Some validators can't be created at field initialization time.
622 This method provides a way to delay their creation until required.
623 """
624 return [*self.default_validators, *self._validators]
625
626 def run_validators(self, value: Any) -> None:
627 if value in self.empty_values:
628 return
629
630 errors = []
631 for v in self.validators:
632 try:
633 v(value)
634 except exceptions.ValidationError as e:
635 if hasattr(e, "code") and e.code in self.error_messages:
636 e.message = self.error_messages[e.code]
637 errors.extend(e.error_list)
638
639 if errors:
640 raise exceptions.ValidationError(errors)
641
642 def validate(self, value: Any, model_instance: Any) -> None:
643 """
644 Validate value and raise ValidationError if necessary. Subclasses
645 should override this to provide validation logic.
646 """
647
648 if self.choices is not None and value not in self.empty_values:
649 for option_key, option_value in self.choices:
650 if isinstance(option_value, list | tuple):
651 # This is an optgroup, so look inside the group for
652 # options.
653 for optgroup_key, optgroup_value in option_value:
654 if value == optgroup_key:
655 return
656 elif value == option_key:
657 return
658 raise exceptions.ValidationError(
659 self.error_messages["invalid_choice"],
660 code="invalid_choice",
661 params={"value": value},
662 )
663
664 if value is None and not self.allow_null:
665 raise exceptions.ValidationError(
666 self.error_messages["allow_null"], code="allow_null"
667 )
668
669 if self.required and value in self.empty_values:
670 raise exceptions.ValidationError(
671 self.error_messages["required"], code="required"
672 )
673
674 def clean(self, value: Any, model_instance: Any) -> Any:
675 """
676 Convert the value's type and run validation. Validation errors
677 from to_python() and validate() are propagated. Return the correct
678 value if no error is raised.
679 """
680 value = self.to_python(value)
681 self.validate(value, model_instance)
682 self.run_validators(value)
683 return value
684
685 def db_type_parameters(self, connection: BaseDatabaseWrapper) -> DictWrapper:
686 return DictWrapper(self.__dict__, connection.ops.quote_name, "qn_")
687
688 def db_check(self, connection: BaseDatabaseWrapper) -> str | None:
689 """
690 Return the database column check constraint for this field, for the
691 provided connection. Works the same way as db_type() for the case that
692 get_internal_type() does not map to a preexisting model field.
693 """
694 data = self.db_type_parameters(connection)
695 try:
696 return (
697 connection.data_type_check_constraints[self.get_internal_type()] % data
698 )
699 except KeyError:
700 return None
701
702 def db_type(self, connection: BaseDatabaseWrapper) -> str | None:
703 """
704 Return the database column data type for this field, for the provided
705 connection.
706 """
707 # The default implementation of this method looks at the
708 # backend-specific data_types dictionary, looking up the field by its
709 # "internal type".
710 #
711 # A Field class can implement the get_internal_type() method to specify
712 # which *preexisting* Plain Field class it's most similar to -- i.e.,
713 # a custom field might be represented by a TEXT column type, which is
714 # the same as the TextField Plain field type, which means the custom
715 # field's get_internal_type() returns 'TextField'.
716 #
717 # But the limitation of the get_internal_type() / data_types approach
718 # is that it cannot handle database column types that aren't already
719 # mapped to one of the built-in Plain field types. In this case, you
720 # can implement db_type() instead of get_internal_type() to specify
721 # exactly which wacky database column type you want to use.
722 data = self.db_type_parameters(connection)
723 try:
724 column_type = connection.data_types[self.get_internal_type()]
725 except KeyError:
726 return None
727 else:
728 # column_type is either a single-parameter function or a string.
729 if callable(column_type):
730 return column_type(data)
731 return column_type % data
732
733 def rel_db_type(self, connection: BaseDatabaseWrapper) -> str | None:
734 """
735 Return the data type that a related field pointing to this field should
736 use. For example, this method is called by ForeignKey to determine its data type.
737 """
738 return self.db_type(connection)
739
740 def cast_db_type(self, connection: BaseDatabaseWrapper) -> str | None:
741 """Return the data type to use in the Cast() function."""
742 db_type = connection.ops.cast_data_types.get(self.get_internal_type())
743 if db_type:
744 return db_type % self.db_type_parameters(connection)
745 return self.db_type(connection)
746
747 def db_parameters(self, connection: BaseDatabaseWrapper) -> dict[str, Any]:
748 """
749 Extension of db_type(), providing a range of different return values
750 (type, checks). This will look at db_type(), allowing custom model
751 fields to override it.
752 """
753 type_string = self.db_type(connection)
754 check_string = self.db_check(connection)
755 return {
756 "type": type_string,
757 "check": check_string,
758 }
759
760 def db_type_suffix(self, connection: BaseDatabaseWrapper) -> str | None:
761 return connection.data_types_suffix.get(self.get_internal_type())
762
763 def get_db_converters(
764 self, connection: BaseDatabaseWrapper
765 ) -> list[Callable[..., Any]]:
766 if hasattr(self, "from_db_value"):
767 return [self.from_db_value]
768 return []
769
770 @property
771 def db_returning(self) -> bool:
772 """
773 Private API intended only to be used by Plain itself. Currently only
774 the PostgreSQL backend supports returning multiple fields on a model.
775 """
776 return False
777
778 def set_attributes_from_name(self, name: str) -> None:
779 self.name = self.name or name
780 self.attname, self.column = self.get_attname_column()
781 self.concrete = self.column is not None
782
783 def contribute_to_class(self, cls: Any, name: str) -> None:
784 """
785 Register the field with the model class it belongs to.
786 """
787 self.set_attributes_from_name(name)
788 self.model = cls
789 cls._model_meta.add_field(self)
790 if self.column:
791 setattr(cls, self.attname, self.descriptor_class(self))
792
793 def get_attname(self) -> str:
794 return self.name
795
796 def get_attname_column(self) -> tuple[str, str]:
797 attname = self.get_attname()
798 column = self.db_column or attname
799 return attname, column
800
801 def get_internal_type(self) -> str:
802 return self.__class__.__name__
803
804 def pre_save(self, model_instance: Any, add: bool) -> Any:
805 """Return field's value just before saving."""
806 return getattr(model_instance, self.attname)
807
808 def get_prep_value(self, value: Any) -> Any:
809 """Perform preliminary non-db specific value checks and conversions."""
810 if isinstance(value, Promise):
811 value = value._proxy____cast()
812 return value
813
814 def get_db_prep_value(
815 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
816 ) -> Any:
817 """
818 Return field's value prepared for interacting with the database backend.
819
820 Used by the default implementations of get_db_prep_save().
821 """
822 if not prepared:
823 value = self.get_prep_value(value)
824 return value
825
826 def get_db_prep_save(self, value: Any, connection: BaseDatabaseWrapper) -> Any:
827 """Return field's value prepared for saving into a database."""
828 if hasattr(value, "as_sql"):
829 return value
830 return self.get_db_prep_value(value, connection=connection, prepared=False)
831
832 def has_default(self) -> bool:
833 """Return a boolean of whether this field has a default value."""
834 return self.default is not NOT_PROVIDED
835
836 def get_default(self) -> Any:
837 """Return the default value for this field."""
838 return self._get_default()
839
840 @cached_property
841 def _get_default(self) -> Callable[[], Any]:
842 if self.has_default():
843 if callable(self.default):
844 return self.default
845 return lambda: self.default
846
847 if not self.empty_strings_allowed or self.allow_null:
848 return return_None
849 return str # return empty string
850
851 def get_choices(
852 self,
853 include_blank: bool = True,
854 blank_choice: list[tuple[str, str]] = BLANK_CHOICE_DASH,
855 limit_choices_to: Any = None,
856 ordering: tuple[str, ...] = (),
857 ) -> list[tuple[Any, str]]:
858 """
859 Return choices with a default blank choices included, for use
860 as <select> choices for this field.
861 """
862 if self.choices is not None:
863 choices = list(self.choices)
864 if include_blank:
865 blank_defined = any(
866 choice in ("", None) for choice, _ in self.flatchoices
867 )
868 if not blank_defined:
869 choices = blank_choice + choices
870 return choices
871 rel_model = self.remote_field.model
872 limit_choices_to = limit_choices_to or self.get_limit_choices_to()
873 choice_func = operator.attrgetter(
874 self.remote_field.get_related_field().attname
875 if hasattr(self.remote_field, "get_related_field")
876 else "id"
877 )
878 qs = rel_model.query.complex_filter(limit_choices_to)
879 if ordering:
880 qs = qs.order_by(*ordering)
881 return (blank_choice if include_blank else []) + [
882 (choice_func(x), str(x)) for x in qs
883 ]
884
885 def value_to_string(self, obj: Any) -> str:
886 """
887 Return a string value of this field from the passed obj.
888 This is used by the serialization framework.
889 """
890 return str(self.value_from_object(obj))
891
892 def _get_flatchoices(self) -> list[tuple[Any, Any]]:
893 """Flattened version of choices tuple."""
894 if self.choices is None:
895 return []
896 flat = []
897 for choice, value in self.choices:
898 if isinstance(value, list | tuple):
899 flat.extend(value)
900 else:
901 flat.append((choice, value))
902 return flat
903
904 flatchoices = property(_get_flatchoices)
905
906 def save_form_data(self, instance: Any, data: Any) -> None:
907 setattr(instance, self.name, data)
908
909 def value_from_object(self, obj: Any) -> Any:
910 """Return the value of this field in the given model instance."""
911 return getattr(obj, self.attname)
912
913
914class BooleanField(Field):
915 empty_strings_allowed = False
916 default_error_messages = {
917 "invalid": "โ%(value)sโ value must be either True or False.",
918 "invalid_nullable": "โ%(value)sโ value must be either True, False, or None.",
919 }
920 description = "Boolean (Either True or False)"
921
922 def get_internal_type(self) -> str:
923 return "BooleanField"
924
925 def to_python(self, value: Any) -> Any:
926 if self.allow_null and value in self.empty_values:
927 return None
928 if value in (True, False):
929 # 1/0 are equal to True/False. bool() converts former to latter.
930 return bool(value)
931 if value in ("t", "True", "1"):
932 return True
933 if value in ("f", "False", "0"):
934 return False
935 raise exceptions.ValidationError(
936 self.error_messages["invalid_nullable" if self.allow_null else "invalid"],
937 code="invalid",
938 params={"value": value},
939 )
940
941 def get_prep_value(self, value: Any) -> Any:
942 value = super().get_prep_value(value)
943 if value is None:
944 return None
945 return self.to_python(value)
946
947
948class CharField(Field):
949 def __init__(self, *, db_collation: str | None = None, **kwargs: Any):
950 super().__init__(**kwargs)
951 self.db_collation = db_collation
952 if self.max_length is not None:
953 self.validators.append(validators.MaxLengthValidator(self.max_length))
954
955 @property
956 def description(self) -> str:
957 if self.max_length is not None:
958 return "String (up to %(max_length)s)"
959 else:
960 return "String (unlimited)"
961
962 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
963 return [
964 *super().preflight(**kwargs),
965 *self._check_db_collation(),
966 *self._check_max_length_attribute(),
967 ]
968
969 def _check_max_length_attribute(self, **kwargs: Any) -> list[PreflightResult]:
970 if self.max_length is None:
971 if (
972 db_connection.features.supports_unlimited_charfield
973 or "supports_unlimited_charfield"
974 in self.model.model_options.required_db_features
975 ):
976 return []
977 return [
978 PreflightResult(
979 fix="CharFields must define a 'max_length' attribute.",
980 obj=self,
981 id="fields.charfield_missing_max_length",
982 )
983 ]
984 elif (
985 not isinstance(self.max_length, int)
986 or isinstance(self.max_length, bool)
987 or self.max_length <= 0
988 ):
989 return [
990 PreflightResult(
991 fix="'max_length' must be a positive integer.",
992 obj=self,
993 id="fields.charfield_invalid_max_length",
994 )
995 ]
996 else:
997 return []
998
999 def _check_db_collation(self) -> list[PreflightResult]:
1000 errors = []
1001 if not (
1002 self.db_collation is None
1003 or "supports_collation_on_charfield"
1004 in self.model.model_options.required_db_features
1005 or db_connection.features.supports_collation_on_charfield
1006 ):
1007 errors.append(
1008 PreflightResult(
1009 fix=f"{db_connection.display_name} does not support a database collation on "
1010 "CharFields.",
1011 obj=self,
1012 id="fields.db_collation_unsupported",
1013 ),
1014 )
1015 return errors
1016
1017 def cast_db_type(self, connection: BaseDatabaseWrapper) -> str | None:
1018 if self.max_length is None:
1019 return connection.ops.cast_char_field_without_max_length
1020 return super().cast_db_type(connection)
1021
1022 def db_parameters(self, connection: BaseDatabaseWrapper) -> dict[str, Any]:
1023 db_params = super().db_parameters(connection)
1024 db_params["collation"] = self.db_collation
1025 return db_params
1026
1027 def get_internal_type(self) -> str:
1028 return "CharField"
1029
1030 def to_python(self, value: Any) -> Any:
1031 if isinstance(value, str) or value is None:
1032 return value
1033 return str(value)
1034
1035 def get_prep_value(self, value: Any) -> Any:
1036 value = super().get_prep_value(value)
1037 return self.to_python(value)
1038
1039 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
1040 name, path, args, kwargs = super().deconstruct()
1041 if self.db_collation:
1042 kwargs["db_collation"] = self.db_collation
1043 return name, path, args, kwargs
1044
1045
1046def _to_naive(value: datetime.datetime) -> datetime.datetime:
1047 if timezone.is_aware(value):
1048 value = timezone.make_naive(value, datetime.UTC)
1049 return value
1050
1051
1052def _get_naive_now() -> datetime.datetime:
1053 return _to_naive(timezone.now())
1054
1055
1056class DateTimeCheckMixin:
1057 def preflight(self, **kwargs: Any) -> list[PreflightResult]: # type: ignore[misc]
1058 return [
1059 *super().preflight(**kwargs), # type: ignore[misc]
1060 *self._check_mutually_exclusive_options(),
1061 *self._check_fix_default_value(),
1062 ]
1063
1064 def _check_mutually_exclusive_options(self) -> list[PreflightResult]:
1065 # auto_now, auto_now_add, and default are mutually exclusive
1066 # options. The use of more than one of these options together
1067 # will trigger an Error
1068 mutually_exclusive_options = [
1069 self.auto_now_add, # type: ignore[attr-defined]
1070 self.auto_now, # type: ignore[attr-defined]
1071 self.has_default(), # type: ignore[attr-defined]
1072 ]
1073 enabled_options = [
1074 option not in (None, False) for option in mutually_exclusive_options
1075 ].count(True)
1076 if enabled_options > 1:
1077 return [
1078 PreflightResult(
1079 fix="The options auto_now, auto_now_add, and default "
1080 "are mutually exclusive. Only one of these options "
1081 "may be present.",
1082 obj=self,
1083 id="fields.datetime_auto_options_mutually_exclusive",
1084 )
1085 ]
1086 else:
1087 return []
1088
1089 def _check_fix_default_value(self) -> list[PreflightResult]:
1090 return []
1091
1092 # Concrete subclasses use this in their implementations of
1093 # _check_fix_default_value().
1094 def _check_if_value_fixed(
1095 self,
1096 value: datetime.date | datetime.datetime,
1097 now: datetime.datetime | None = None,
1098 ) -> list[PreflightResult]:
1099 """
1100 Check if the given value appears to have been provided as a "fixed"
1101 time value, and include a warning in the returned list if it does. The
1102 value argument must be a date object or aware/naive datetime object. If
1103 now is provided, it must be a naive datetime object.
1104 """
1105 if now is None:
1106 now = _get_naive_now()
1107 offset = datetime.timedelta(seconds=10)
1108 lower = now - offset
1109 upper = now + offset
1110 if isinstance(value, datetime.datetime):
1111 value = _to_naive(value)
1112 else:
1113 assert isinstance(value, datetime.date)
1114 lower = lower.date()
1115 upper = upper.date()
1116 if lower <= value <= upper:
1117 return [
1118 PreflightResult(
1119 fix="Fixed default value provided. "
1120 "It seems you set a fixed date / time / datetime "
1121 "value as default for this field. This may not be "
1122 "what you want. If you want to have the current date "
1123 "as default, use `plain.utils.timezone.now`",
1124 obj=self,
1125 id="fields.datetime_naive_default_value",
1126 warning=True,
1127 )
1128 ]
1129 return []
1130
1131
1132class DateField(DateTimeCheckMixin, Field):
1133 empty_strings_allowed = False
1134 default_error_messages = {
1135 "invalid": "โ%(value)sโ value has an invalid date format. It must be in YYYY-MM-DD format.",
1136 "invalid_date": "โ%(value)sโ value has the correct format (YYYY-MM-DD) but it is an invalid date.",
1137 }
1138 description = "Date (without time)"
1139
1140 def __init__(
1141 self, *, auto_now: bool = False, auto_now_add: bool = False, **kwargs: Any
1142 ):
1143 self.auto_now, self.auto_now_add = auto_now, auto_now_add
1144 if auto_now or auto_now_add:
1145 kwargs["required"] = False
1146 super().__init__(**kwargs)
1147
1148 def _check_fix_default_value(self) -> list[PreflightResult]:
1149 """
1150 Warn that using an actual date or datetime value is probably wrong;
1151 it's only evaluated on server startup.
1152 """
1153 if not self.has_default():
1154 return []
1155
1156 value = self.default
1157 if isinstance(value, datetime.datetime):
1158 value = _to_naive(value).date()
1159 elif isinstance(value, datetime.date):
1160 pass
1161 else:
1162 # No explicit date / datetime value -- no checks necessary
1163 return []
1164 # At this point, value is a date object.
1165 return self._check_if_value_fixed(value)
1166
1167 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
1168 name, path, args, kwargs = super().deconstruct()
1169 if self.auto_now:
1170 kwargs["auto_now"] = True
1171 if self.auto_now_add:
1172 kwargs["auto_now_add"] = True
1173 if self.auto_now or self.auto_now_add:
1174 del kwargs["required"]
1175 return name, path, args, kwargs
1176
1177 def get_internal_type(self) -> str:
1178 return "DateField"
1179
1180 def to_python(self, value: Any) -> Any:
1181 if value is None:
1182 return value
1183 if isinstance(value, datetime.datetime):
1184 if timezone.is_aware(value):
1185 # Convert aware datetimes to the default time zone
1186 # before casting them to dates (#17742).
1187 default_timezone = timezone.get_default_timezone()
1188 value = timezone.make_naive(value, default_timezone)
1189 return value.date()
1190 if isinstance(value, datetime.date):
1191 return value
1192
1193 try:
1194 parsed = parse_date(value)
1195 if parsed is not None:
1196 return parsed
1197 except ValueError:
1198 raise exceptions.ValidationError(
1199 self.error_messages["invalid_date"],
1200 code="invalid_date",
1201 params={"value": value},
1202 )
1203
1204 raise exceptions.ValidationError(
1205 self.error_messages["invalid"],
1206 code="invalid",
1207 params={"value": value},
1208 )
1209
1210 def pre_save(self, model_instance: Any, add: bool) -> Any:
1211 if self.auto_now or (self.auto_now_add and add):
1212 value = datetime.date.today()
1213 setattr(model_instance, self.attname, value)
1214 return value
1215 else:
1216 return super().pre_save(model_instance, add)
1217
1218 def get_prep_value(self, value: Any) -> Any:
1219 value = super().get_prep_value(value)
1220 return self.to_python(value)
1221
1222 def get_db_prep_value(
1223 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
1224 ) -> Any:
1225 # Casts dates into the format expected by the backend
1226 if not prepared:
1227 value = self.get_prep_value(value)
1228 return connection.ops.adapt_datefield_value(value)
1229
1230 def value_to_string(self, obj: Any) -> str:
1231 val = self.value_from_object(obj)
1232 return "" if val is None else val.isoformat()
1233
1234
1235class DateTimeField(DateField):
1236 empty_strings_allowed = False
1237 default_error_messages = {
1238 "invalid": "โ%(value)sโ value has an invalid format. It must be in YYYY-MM-DD HH:MM[:ss[.uuuuuu]][TZ] format.",
1239 "invalid_date": "โ%(value)sโ value has the correct format (YYYY-MM-DD) but it is an invalid date.",
1240 "invalid_datetime": "โ%(value)sโ value has the correct format (YYYY-MM-DD HH:MM[:ss[.uuuuuu]][TZ]) but it is an invalid date/time.",
1241 }
1242 description = "Date (with time)"
1243
1244 # __init__ is inherited from DateField
1245
1246 def _check_fix_default_value(self) -> list[PreflightResult]:
1247 """
1248 Warn that using an actual date or datetime value is probably wrong;
1249 it's only evaluated on server startup.
1250 """
1251 if not self.has_default():
1252 return []
1253
1254 value = self.default
1255 if isinstance(value, datetime.datetime | datetime.date):
1256 return self._check_if_value_fixed(value)
1257 # No explicit date / datetime value -- no checks necessary.
1258 return []
1259
1260 def get_internal_type(self) -> str:
1261 return "DateTimeField"
1262
1263 def to_python(self, value: Any) -> Any:
1264 if value is None:
1265 return value
1266 if isinstance(value, datetime.datetime):
1267 return value
1268 if isinstance(value, datetime.date):
1269 value = datetime.datetime(value.year, value.month, value.day)
1270
1271 # For backwards compatibility, interpret naive datetimes in
1272 # local time. This won't work during DST change, but we can't
1273 # do much about it, so we let the exceptions percolate up the
1274 # call stack.
1275 warnings.warn(
1276 f"DateTimeField {self.model.__name__}.{self.name} received a naive datetime "
1277 f"({value}) while time zone support is active.",
1278 RuntimeWarning,
1279 )
1280 default_timezone = timezone.get_default_timezone()
1281 value = timezone.make_aware(value, default_timezone)
1282
1283 return value
1284
1285 try:
1286 parsed = parse_datetime(value)
1287 if parsed is not None:
1288 return parsed
1289 except ValueError:
1290 raise exceptions.ValidationError(
1291 self.error_messages["invalid_datetime"],
1292 code="invalid_datetime",
1293 params={"value": value},
1294 )
1295
1296 try:
1297 parsed = parse_date(value)
1298 if parsed is not None:
1299 return datetime.datetime(parsed.year, parsed.month, parsed.day)
1300 except ValueError:
1301 raise exceptions.ValidationError(
1302 self.error_messages["invalid_date"],
1303 code="invalid_date",
1304 params={"value": value},
1305 )
1306
1307 raise exceptions.ValidationError(
1308 self.error_messages["invalid"],
1309 code="invalid",
1310 params={"value": value},
1311 )
1312
1313 def pre_save(self, model_instance: Any, add: bool) -> Any:
1314 if self.auto_now or (self.auto_now_add and add):
1315 value = timezone.now()
1316 setattr(model_instance, self.attname, value)
1317 return value
1318 else:
1319 return super().pre_save(model_instance, add)
1320
1321 def get_prep_value(self, value: Any) -> Any:
1322 value = super().get_prep_value(value)
1323 value = self.to_python(value)
1324 if value is not None and timezone.is_naive(value):
1325 # For backwards compatibility, interpret naive datetimes in local
1326 # time. This won't work during DST change, but we can't do much
1327 # about it, so we let the exceptions percolate up the call stack.
1328 try:
1329 name = f"{self.model.__name__}.{self.name}"
1330 except AttributeError:
1331 name = "(unbound)"
1332 warnings.warn(
1333 f"DateTimeField {name} received a naive datetime ({value})"
1334 " while time zone support is active.",
1335 RuntimeWarning,
1336 )
1337 default_timezone = timezone.get_default_timezone()
1338 value = timezone.make_aware(value, default_timezone)
1339 return value
1340
1341 def get_db_prep_value(
1342 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
1343 ) -> Any:
1344 # Casts datetimes into the format expected by the backend
1345 if not prepared:
1346 value = self.get_prep_value(value)
1347 return connection.ops.adapt_datetimefield_value(value)
1348
1349 def value_to_string(self, obj: Any) -> str:
1350 val = self.value_from_object(obj)
1351 return "" if val is None else val.isoformat()
1352
1353
1354class DecimalField(Field):
1355 empty_strings_allowed = False
1356 default_error_messages = {
1357 "invalid": "โ%(value)sโ value must be a decimal number.",
1358 }
1359 description = "Decimal number"
1360
1361 def __init__(
1362 self,
1363 *,
1364 max_digits: int | None = None,
1365 decimal_places: int | None = None,
1366 **kwargs: Any,
1367 ):
1368 self.max_digits, self.decimal_places = max_digits, decimal_places
1369 super().__init__(**kwargs)
1370
1371 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
1372 errors = super().preflight(**kwargs)
1373
1374 digits_errors = [
1375 *self._check_decimal_places(),
1376 *self._check_max_digits(),
1377 ]
1378 if not digits_errors:
1379 errors.extend(self._check_decimal_places_and_max_digits())
1380 else:
1381 errors.extend(digits_errors)
1382 return errors
1383
1384 def _check_decimal_places(self) -> list[PreflightResult]:
1385 try:
1386 decimal_places = int(self.decimal_places)
1387 if decimal_places < 0:
1388 raise ValueError()
1389 except TypeError:
1390 return [
1391 PreflightResult(
1392 fix="DecimalFields must define a 'decimal_places' attribute.",
1393 obj=self,
1394 id="fields.decimalfield_missing_decimal_places",
1395 )
1396 ]
1397 except ValueError:
1398 return [
1399 PreflightResult(
1400 fix="'decimal_places' must be a non-negative integer.",
1401 obj=self,
1402 id="fields.decimalfield_invalid_decimal_places",
1403 )
1404 ]
1405 else:
1406 return []
1407
1408 def _check_max_digits(self) -> list[PreflightResult]:
1409 try:
1410 max_digits = int(self.max_digits)
1411 if max_digits <= 0:
1412 raise ValueError()
1413 except TypeError:
1414 return [
1415 PreflightResult(
1416 fix="DecimalFields must define a 'max_digits' attribute.",
1417 obj=self,
1418 id="fields.decimalfield_missing_max_digits",
1419 )
1420 ]
1421 except ValueError:
1422 return [
1423 PreflightResult(
1424 fix="'max_digits' must be a positive integer.",
1425 obj=self,
1426 id="fields.decimalfield_invalid_max_digits",
1427 )
1428 ]
1429 else:
1430 return []
1431
1432 def _check_decimal_places_and_max_digits(self) -> list[PreflightResult]:
1433 if int(self.decimal_places) > int(self.max_digits):
1434 return [
1435 PreflightResult(
1436 fix="'max_digits' must be greater or equal to 'decimal_places'.",
1437 obj=self,
1438 id="fields.decimalfield_decimal_places_exceeds_max_digits",
1439 )
1440 ]
1441 return []
1442
1443 @cached_property
1444 def validators(self) -> list[Callable[..., Any]]:
1445 return super().validators + [
1446 validators.DecimalValidator(self.max_digits, self.decimal_places)
1447 ]
1448
1449 @cached_property
1450 def context(self) -> decimal.Context:
1451 return decimal.Context(prec=self.max_digits)
1452
1453 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
1454 name, path, args, kwargs = super().deconstruct()
1455 if self.max_digits is not None:
1456 kwargs["max_digits"] = self.max_digits
1457 if self.decimal_places is not None:
1458 kwargs["decimal_places"] = self.decimal_places
1459 return name, path, args, kwargs
1460
1461 def get_internal_type(self) -> str:
1462 return "DecimalField"
1463
1464 def to_python(self, value: Any) -> Any:
1465 if value is None:
1466 return value
1467 try:
1468 if isinstance(value, float):
1469 decimal_value = self.context.create_decimal_from_float(value)
1470 else:
1471 decimal_value = decimal.Decimal(value)
1472 except (decimal.InvalidOperation, TypeError, ValueError):
1473 raise exceptions.ValidationError(
1474 self.error_messages["invalid"],
1475 code="invalid",
1476 params={"value": value},
1477 )
1478 if not decimal_value.is_finite():
1479 raise exceptions.ValidationError(
1480 self.error_messages["invalid"],
1481 code="invalid",
1482 params={"value": value},
1483 )
1484 return decimal_value
1485
1486 def get_db_prep_value(
1487 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
1488 ) -> Any:
1489 if not prepared:
1490 value = self.get_prep_value(value)
1491 if hasattr(value, "as_sql"):
1492 return value
1493 return connection.ops.adapt_decimalfield_value(
1494 value, self.max_digits, self.decimal_places
1495 )
1496
1497 def get_prep_value(self, value: Any) -> Any:
1498 value = super().get_prep_value(value)
1499 return self.to_python(value)
1500
1501
1502class DurationField(Field):
1503 """
1504 Store timedelta objects.
1505
1506 Use interval on PostgreSQL, INTERVAL DAY TO SECOND on Oracle, and bigint
1507 of microseconds on other databases.
1508 """
1509
1510 empty_strings_allowed = False
1511 default_error_messages = {
1512 "invalid": "โ%(value)sโ value has an invalid format. It must be in [DD] [[HH:]MM:]ss[.uuuuuu] format.",
1513 }
1514 description = "Duration"
1515
1516 def get_internal_type(self) -> str:
1517 return "DurationField"
1518
1519 def to_python(self, value: Any) -> Any:
1520 if value is None:
1521 return value
1522 if isinstance(value, datetime.timedelta):
1523 return value
1524 try:
1525 parsed = parse_duration(value)
1526 except ValueError:
1527 pass
1528 else:
1529 if parsed is not None:
1530 return parsed
1531
1532 raise exceptions.ValidationError(
1533 self.error_messages["invalid"],
1534 code="invalid",
1535 params={"value": value},
1536 )
1537
1538 def get_db_prep_value(
1539 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
1540 ) -> Any:
1541 if connection.features.has_native_duration_field:
1542 return value
1543 if value is None:
1544 return None
1545 return duration_microseconds(value)
1546
1547 def get_db_converters(
1548 self, connection: BaseDatabaseWrapper
1549 ) -> list[Callable[..., Any]]:
1550 converters = []
1551 if not connection.features.has_native_duration_field:
1552 converters.append(connection.ops.convert_durationfield_value)
1553 return converters + super().get_db_converters(connection)
1554
1555 def value_to_string(self, obj: Any) -> str:
1556 val = self.value_from_object(obj)
1557 return "" if val is None else duration_string(val)
1558
1559
1560class EmailField(CharField):
1561 default_validators = [validators.validate_email]
1562 description = "Email address"
1563
1564 def __init__(self, **kwargs: Any):
1565 # max_length=254 to be compliant with RFCs 3696 and 5321
1566 kwargs.setdefault("max_length", 254)
1567 super().__init__(**kwargs)
1568
1569 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
1570 name, path, args, kwargs = super().deconstruct()
1571 # We do not exclude max_length if it matches default as we want to change
1572 # the default in future.
1573 return name, path, args, kwargs
1574
1575
1576class FloatField(Field):
1577 empty_strings_allowed = False
1578 default_error_messages = {
1579 "invalid": "โ%(value)sโ value must be a float.",
1580 }
1581 description = "Floating point number"
1582
1583 def get_prep_value(self, value: Any) -> Any:
1584 value = super().get_prep_value(value)
1585 if value is None:
1586 return None
1587 try:
1588 return float(value)
1589 except (TypeError, ValueError) as e:
1590 raise e.__class__(
1591 f"Field '{self.name}' expected a number but got {value!r}.",
1592 ) from e
1593
1594 def get_internal_type(self) -> str:
1595 return "FloatField"
1596
1597 def to_python(self, value: Any) -> Any:
1598 if value is None:
1599 return value
1600 try:
1601 return float(value)
1602 except (TypeError, ValueError):
1603 raise exceptions.ValidationError(
1604 self.error_messages["invalid"],
1605 code="invalid",
1606 params={"value": value},
1607 )
1608
1609
1610class IntegerField(Field):
1611 empty_strings_allowed = False
1612 default_error_messages = {
1613 "invalid": "โ%(value)sโ value must be an integer.",
1614 }
1615 description = "Integer"
1616
1617 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
1618 return [
1619 *super().preflight(**kwargs),
1620 *self._check_max_length_warning(),
1621 ]
1622
1623 def _check_max_length_warning(self) -> list[PreflightResult]:
1624 if self.max_length is not None:
1625 return [
1626 PreflightResult(
1627 fix=f"'max_length' is ignored when used with {self.__class__.__name__}. Remove 'max_length' from field.",
1628 obj=self,
1629 id="fields.max_length_ignored",
1630 warning=True,
1631 )
1632 ]
1633 return []
1634
1635 @cached_property
1636 def validators(self) -> list[Callable[..., Any]]:
1637 # These validators can't be added at field initialization time since
1638 # they're based on values retrieved from the database connection.
1639 validators_ = super().validators
1640 internal_type = self.get_internal_type()
1641 min_value, max_value = db_connection.ops.integer_field_range(internal_type)
1642 if min_value is not None and not any(
1643 (
1644 isinstance(validator, validators.MinValueValidator)
1645 and (
1646 validator.limit_value()
1647 if callable(validator.limit_value)
1648 else validator.limit_value
1649 )
1650 >= min_value
1651 )
1652 for validator in validators_
1653 ):
1654 validators_.append(validators.MinValueValidator(min_value))
1655 if max_value is not None and not any(
1656 (
1657 isinstance(validator, validators.MaxValueValidator)
1658 and (
1659 validator.limit_value()
1660 if callable(validator.limit_value)
1661 else validator.limit_value
1662 )
1663 <= max_value
1664 )
1665 for validator in validators_
1666 ):
1667 validators_.append(validators.MaxValueValidator(max_value))
1668 return validators_
1669
1670 def get_prep_value(self, value: Any) -> Any:
1671 value = super().get_prep_value(value)
1672 if value is None:
1673 return None
1674 try:
1675 return int(value)
1676 except (TypeError, ValueError) as e:
1677 raise e.__class__(
1678 f"Field '{self.name}' expected a number but got {value!r}.",
1679 ) from e
1680
1681 def get_db_prep_value(
1682 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
1683 ) -> Any:
1684 value = super().get_db_prep_value(value, connection, prepared)
1685 return connection.ops.adapt_integerfield_value(value, self.get_internal_type())
1686
1687 def get_internal_type(self) -> str:
1688 return "IntegerField"
1689
1690 def to_python(self, value: Any) -> Any:
1691 if value is None:
1692 return value
1693 try:
1694 return int(value)
1695 except (TypeError, ValueError):
1696 raise exceptions.ValidationError(
1697 self.error_messages["invalid"],
1698 code="invalid",
1699 params={"value": value},
1700 )
1701
1702
1703class BigIntegerField(IntegerField):
1704 description = "Big (8 byte) integer"
1705
1706 def get_internal_type(self) -> str:
1707 return "BigIntegerField"
1708
1709
1710class SmallIntegerField(IntegerField):
1711 description = "Small integer"
1712
1713 def get_internal_type(self) -> str:
1714 return "SmallIntegerField"
1715
1716
1717class GenericIPAddressField(Field):
1718 empty_strings_allowed = False
1719 description = "IP address"
1720 default_error_messages = {}
1721
1722 def __init__(
1723 self,
1724 *,
1725 protocol: str = "both",
1726 unpack_ipv4: bool = False,
1727 **kwargs: Any,
1728 ):
1729 self.unpack_ipv4 = unpack_ipv4
1730 self.protocol = protocol
1731 (
1732 self.default_validators,
1733 invalid_error_message,
1734 ) = validators.ip_address_validators(protocol, unpack_ipv4)
1735 self.default_error_messages["invalid"] = invalid_error_message
1736 kwargs["max_length"] = 39
1737 super().__init__(**kwargs)
1738
1739 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
1740 return [
1741 *super().preflight(**kwargs),
1742 *self._check_required_and_null_values(),
1743 ]
1744
1745 def _check_required_and_null_values(self) -> list[PreflightResult]:
1746 if not getattr(self, "allow_null", False) and not getattr(
1747 self, "required", True
1748 ):
1749 return [
1750 PreflightResult(
1751 fix="GenericIPAddressFields cannot have required=False if allow_null=False, "
1752 "as blank values are stored as nulls.",
1753 obj=self,
1754 id="fields.generic_ip_field_null_blank_config",
1755 )
1756 ]
1757 return []
1758
1759 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
1760 name, path, args, kwargs = super().deconstruct()
1761 if self.unpack_ipv4 is not False:
1762 kwargs["unpack_ipv4"] = self.unpack_ipv4
1763 if self.protocol != "both":
1764 kwargs["protocol"] = self.protocol
1765 if kwargs.get("max_length") == 39:
1766 del kwargs["max_length"]
1767 return name, path, args, kwargs
1768
1769 def get_internal_type(self) -> str:
1770 return "GenericIPAddressField"
1771
1772 def to_python(self, value: Any) -> Any:
1773 if value is None:
1774 return None
1775 if not isinstance(value, str):
1776 value = str(value)
1777 value = value.strip()
1778 if ":" in value:
1779 return clean_ipv6_address(
1780 value, self.unpack_ipv4, self.error_messages["invalid"]
1781 )
1782 return value
1783
1784 def get_db_prep_value(
1785 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
1786 ) -> Any:
1787 if not prepared:
1788 value = self.get_prep_value(value)
1789 return connection.ops.adapt_ipaddressfield_value(value)
1790
1791 def get_prep_value(self, value: Any) -> Any:
1792 value = super().get_prep_value(value)
1793 if value is None:
1794 return None
1795 if value and ":" in value:
1796 try:
1797 return clean_ipv6_address(value, self.unpack_ipv4)
1798 except exceptions.ValidationError:
1799 pass
1800 return str(value)
1801
1802
1803class PositiveIntegerRelDbTypeMixin:
1804 def __init_subclass__(cls, **kwargs: Any) -> None:
1805 super().__init_subclass__(**kwargs)
1806 if not hasattr(cls, "integer_field_class"):
1807 cls.integer_field_class = next(
1808 (
1809 parent
1810 for parent in cls.__mro__[1:]
1811 if issubclass(parent, IntegerField)
1812 ),
1813 None,
1814 )
1815
1816 def rel_db_type(self, connection: BaseDatabaseWrapper) -> str | None:
1817 """
1818 Return the data type that a related field pointing to this field should
1819 use. In most cases, a foreign key pointing to a positive integer
1820 primary key will have an integer column data type but some databases
1821 (e.g. MySQL) have an unsigned integer type. In that case
1822 (related_fields_match_type=True), the primary key should return its
1823 db_type.
1824 """
1825 if connection.features.related_fields_match_type:
1826 return self.db_type(connection)
1827 else:
1828 return self.integer_field_class().db_type(connection=connection)
1829
1830
1831class PositiveBigIntegerField(PositiveIntegerRelDbTypeMixin, BigIntegerField):
1832 description = "Positive big integer"
1833
1834 def get_internal_type(self) -> str:
1835 return "PositiveBigIntegerField"
1836
1837
1838class PositiveIntegerField(PositiveIntegerRelDbTypeMixin, IntegerField):
1839 description = "Positive integer"
1840
1841 def get_internal_type(self) -> str:
1842 return "PositiveIntegerField"
1843
1844
1845class PositiveSmallIntegerField(PositiveIntegerRelDbTypeMixin, SmallIntegerField):
1846 description = "Positive small integer"
1847
1848 def get_internal_type(self) -> str:
1849 return "PositiveSmallIntegerField"
1850
1851
1852class TextField(Field):
1853 description = "Text"
1854
1855 def __init__(self, *, db_collation: str | None = None, **kwargs: Any):
1856 super().__init__(**kwargs)
1857 self.db_collation = db_collation
1858
1859 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
1860 return [
1861 *super().preflight(**kwargs),
1862 *self._check_db_collation(),
1863 ]
1864
1865 def _check_db_collation(self) -> list[PreflightResult]:
1866 errors = []
1867 if not (
1868 self.db_collation is None
1869 or "supports_collation_on_textfield"
1870 in self.model.model_options.required_db_features
1871 or db_connection.features.supports_collation_on_textfield
1872 ):
1873 errors.append(
1874 PreflightResult(
1875 fix=f"{db_connection.display_name} does not support a database collation on "
1876 "TextFields.",
1877 obj=self,
1878 id="fields.db_collation_unsupported",
1879 ),
1880 )
1881 return errors
1882
1883 def db_parameters(self, connection: BaseDatabaseWrapper) -> dict[str, Any]:
1884 db_params = super().db_parameters(connection)
1885 db_params["collation"] = self.db_collation
1886 return db_params
1887
1888 def get_internal_type(self) -> str:
1889 return "TextField"
1890
1891 def to_python(self, value: Any) -> Any:
1892 if isinstance(value, str) or value is None:
1893 return value
1894 return str(value)
1895
1896 def get_prep_value(self, value: Any) -> Any:
1897 value = super().get_prep_value(value)
1898 return self.to_python(value)
1899
1900 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
1901 name, path, args, kwargs = super().deconstruct()
1902 if self.db_collation:
1903 kwargs["db_collation"] = self.db_collation
1904 return name, path, args, kwargs
1905
1906
1907class TimeField(DateTimeCheckMixin, Field):
1908 empty_strings_allowed = False
1909 default_error_messages = {
1910 "invalid": "โ%(value)sโ value has an invalid format. It must be in HH:MM[:ss[.uuuuuu]] format.",
1911 "invalid_time": "โ%(value)sโ value has the correct format (HH:MM[:ss[.uuuuuu]]) but it is an invalid time.",
1912 }
1913 description = "Time"
1914
1915 def __init__(
1916 self, *, auto_now: bool = False, auto_now_add: bool = False, **kwargs: Any
1917 ):
1918 self.auto_now, self.auto_now_add = auto_now, auto_now_add
1919 if auto_now or auto_now_add:
1920 kwargs["required"] = False
1921 super().__init__(**kwargs)
1922
1923 def _check_fix_default_value(self) -> list[PreflightResult]:
1924 """
1925 Warn that using an actual date or datetime value is probably wrong;
1926 it's only evaluated on server startup.
1927 """
1928 if not self.has_default():
1929 return []
1930
1931 value = self.default
1932 if isinstance(value, datetime.datetime):
1933 now = None
1934 elif isinstance(value, datetime.time):
1935 now = _get_naive_now()
1936 # This will not use the right date in the race condition where now
1937 # is just before the date change and value is just past 0:00.
1938 value = datetime.datetime.combine(now.date(), value)
1939 else:
1940 # No explicit time / datetime value -- no checks necessary
1941 return []
1942 # At this point, value is a datetime object.
1943 return self._check_if_value_fixed(value, now=now)
1944
1945 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
1946 name, path, args, kwargs = super().deconstruct()
1947 if self.auto_now is not False:
1948 kwargs["auto_now"] = self.auto_now
1949 if self.auto_now_add is not False:
1950 kwargs["auto_now_add"] = self.auto_now_add
1951 if self.auto_now or self.auto_now_add:
1952 del kwargs["required"]
1953 return name, path, args, kwargs
1954
1955 def get_internal_type(self) -> str:
1956 return "TimeField"
1957
1958 def to_python(self, value: Any) -> Any:
1959 if value is None:
1960 return None
1961 if isinstance(value, datetime.time):
1962 return value
1963 if isinstance(value, datetime.datetime):
1964 # Not usually a good idea to pass in a datetime here (it loses
1965 # information), but this can be a side-effect of interacting with a
1966 # database backend (e.g. Oracle), so we'll be accommodating.
1967 return value.time()
1968
1969 try:
1970 parsed = parse_time(value)
1971 if parsed is not None:
1972 return parsed
1973 except ValueError:
1974 raise exceptions.ValidationError(
1975 self.error_messages["invalid_time"],
1976 code="invalid_time",
1977 params={"value": value},
1978 )
1979
1980 raise exceptions.ValidationError(
1981 self.error_messages["invalid"],
1982 code="invalid",
1983 params={"value": value},
1984 )
1985
1986 def pre_save(self, model_instance: Any, add: bool) -> Any:
1987 if self.auto_now or (self.auto_now_add and add):
1988 value = datetime.datetime.now().time()
1989 setattr(model_instance, self.attname, value)
1990 return value
1991 else:
1992 return super().pre_save(model_instance, add)
1993
1994 def get_prep_value(self, value: Any) -> Any:
1995 value = super().get_prep_value(value)
1996 return self.to_python(value)
1997
1998 def get_db_prep_value(
1999 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
2000 ) -> Any:
2001 # Casts times into the format expected by the backend
2002 if not prepared:
2003 value = self.get_prep_value(value)
2004 return connection.ops.adapt_timefield_value(value)
2005
2006 def value_to_string(self, obj: Any) -> str:
2007 val = self.value_from_object(obj)
2008 return "" if val is None else val.isoformat()
2009
2010
2011class URLField(CharField):
2012 default_validators = [validators.URLValidator()]
2013 description = "URL"
2014
2015 def __init__(self, **kwargs: Any):
2016 kwargs.setdefault("max_length", 200)
2017 super().__init__(**kwargs)
2018
2019 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
2020 name, path, args, kwargs = super().deconstruct()
2021 if kwargs.get("max_length") == 200:
2022 del kwargs["max_length"]
2023 return name, path, args, kwargs
2024
2025
2026class BinaryField(Field):
2027 description = "Raw binary data"
2028 empty_values = [None, b""]
2029
2030 def __init__(self, **kwargs: Any):
2031 super().__init__(**kwargs)
2032 if self.max_length is not None:
2033 self.validators.append(validators.MaxLengthValidator(self.max_length))
2034
2035 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
2036 return [*super().preflight(**kwargs), *self._check_str_default_value()]
2037
2038 def _check_str_default_value(self) -> list[PreflightResult]:
2039 if self.has_default() and isinstance(self.default, str):
2040 return [
2041 PreflightResult(
2042 fix="BinaryField's default cannot be a string. Use bytes "
2043 "content instead.",
2044 obj=self,
2045 id="fields.filefield_upload_to_not_callable",
2046 )
2047 ]
2048 return []
2049
2050 def get_internal_type(self) -> str:
2051 return "BinaryField"
2052
2053 def get_placeholder(
2054 self, value: Any, compiler: SQLCompiler, connection: BaseDatabaseWrapper
2055 ) -> Any:
2056 return connection.ops.binary_placeholder_sql(value)
2057
2058 def get_default(self) -> Any:
2059 if self.has_default() and not callable(self.default):
2060 return self.default
2061 default = super().get_default()
2062 if default == "":
2063 return b""
2064 return default
2065
2066 def get_db_prep_value(
2067 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
2068 ) -> Any:
2069 value = super().get_db_prep_value(value, connection, prepared)
2070 if value is not None:
2071 return connection.Database.Binary(value) # type: ignore[attr-defined]
2072 return value
2073
2074 def value_to_string(self, obj: Any) -> str:
2075 """Binary data is serialized as base64"""
2076 return b64encode(self.value_from_object(obj)).decode("ascii")
2077
2078 def to_python(self, value: Any) -> Any:
2079 # If it's a string, it should be base64-encoded data
2080 if isinstance(value, str):
2081 return memoryview(b64decode(value.encode("ascii")))
2082 return value
2083
2084
2085class UUIDField(Field):
2086 default_error_messages = {
2087 "invalid": "โ%(value)sโ is not a valid UUID.",
2088 }
2089 description = "Universally unique identifier"
2090 empty_strings_allowed = False
2091
2092 def __init__(self, **kwargs: Any):
2093 kwargs["max_length"] = 32
2094 super().__init__(**kwargs)
2095
2096 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
2097 name, path, args, kwargs = super().deconstruct()
2098 del kwargs["max_length"]
2099 return name, path, args, kwargs
2100
2101 def get_internal_type(self) -> str:
2102 return "UUIDField"
2103
2104 def get_prep_value(self, value: Any) -> Any:
2105 value = super().get_prep_value(value)
2106 return self.to_python(value)
2107
2108 def get_db_prep_value(
2109 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
2110 ) -> Any:
2111 if value is None:
2112 return None
2113 if not isinstance(value, uuid.UUID):
2114 value = self.to_python(value)
2115
2116 if connection.features.has_native_uuid_field:
2117 return value
2118 return value.hex
2119
2120 def to_python(self, value: Any) -> Any:
2121 if value is not None and not isinstance(value, uuid.UUID):
2122 input_form = "int" if isinstance(value, int) else "hex"
2123 try:
2124 return uuid.UUID(**{input_form: value})
2125 except (AttributeError, ValueError):
2126 raise exceptions.ValidationError(
2127 self.error_messages["invalid"],
2128 code="invalid",
2129 params={"value": value},
2130 )
2131 return value
2132
2133
2134class PrimaryKeyField(BigIntegerField):
2135 db_returning = True
2136
2137 def __init__(self):
2138 super().__init__(required=False)
2139 self.primary_key = True
2140 self.auto_created = True
2141 # Adjust creation counter for auto-created fields
2142 # We need to undo the counter increment from Field.__init__ and use the auto counter
2143 Field.creation_counter -= 1 # Undo the increment
2144 self.creation_counter = Field.auto_creation_counter
2145 Field.auto_creation_counter -= 1
2146
2147 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
2148 errors = super().preflight(**kwargs)
2149 # Remove the reserved_field_name_id error for 'id' field name since PrimaryKeyField is allowed to use it
2150 errors = [e for e in errors if e.id != "fields.reserved_field_name_id"]
2151 return errors
2152
2153 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
2154 # PrimaryKeyField takes no parameters, so we return an empty kwargs dict
2155 return (self.name, "plain.models.PrimaryKeyField", [], {})
2156
2157 def validate(self, value: Any, model_instance: Any) -> None:
2158 pass
2159
2160 def get_db_prep_value(
2161 self, value: Any, connection: BaseDatabaseWrapper, prepared: bool = False
2162 ) -> Any:
2163 if not prepared:
2164 value = self.get_prep_value(value)
2165 value = connection.ops.validate_autopk_value(value)
2166 return value
2167
2168 def get_internal_type(self) -> str:
2169 return "PrimaryKeyField"
2170
2171 def rel_db_type(self, connection: BaseDatabaseWrapper) -> str | None:
2172 return BigIntegerField().db_type(connection=connection)