1from __future__ import annotations
2
3import copy
4import warnings
5from collections.abc import Collection, Iterable, Iterator, Sequence
6from contextlib import contextmanager
7from itertools import chain
8from typing import TYPE_CHECKING, Any, Self
9
10if TYPE_CHECKING:
11 from plain.postgres.meta import Meta
12 from plain.postgres.options import Options
13
14import plain.runtime
15import psycopg
16from plain.exceptions import ValidationError
17from plain.postgres import models_registry, transaction, types
18from plain.postgres.constants import LOOKUP_SEP
19from plain.postgres.constraints import CheckConstraint, UniqueConstraint
20from plain.postgres.db import PLAIN_VERSION_PICKLE_KEY
21from plain.postgres.dialect import MAX_NAME_LENGTH
22from plain.postgres.exceptions import (
23 DoesNotExistDescriptor,
24 FieldDoesNotExist,
25 FieldError,
26 MultipleObjectsReturnedDescriptor,
27)
28from plain.postgres.expressions import RawSQL, Value
29from plain.postgres.fields import DATABASE_DEFAULT, Field
30from plain.postgres.fields.related import RelatedField
31from plain.postgres.fields.reverse_related import ForeignObjectRel
32from plain.postgres.meta import Meta
33from plain.postgres.options import Options
34from plain.postgres.query import F, Q, QuerySet
35from plain.preflight import PreflightResult
36from plain.utils.encoding import force_str
37from plain.utils.hashable import make_hashable
38
39
40class Deferred:
41 def __repr__(self) -> str:
42 return "<Deferred field>"
43
44 def __str__(self) -> str:
45 return "<Deferred field>"
46
47
48DEFERRED = Deferred()
49
50
51class ModelBase(type):
52 """Metaclass for all models."""
53
54 def __new__(
55 cls, name: str, bases: tuple[type, ...], attrs: dict[str, Any], **kwargs: Any
56 ) -> type:
57 # Don't do any of this for the root models.Model class.
58 if not bases:
59 return super().__new__(cls, name, bases, attrs)
60
61 for base in bases:
62 # Models are required to directly inherit from model.Model, not a subclass of it.
63 if issubclass(base, Model) and base is not Model:
64 raise TypeError(
65 f"A model can't extend another model: {name} extends {base}"
66 )
67
68 return super().__new__(cls, name, bases, attrs, **kwargs)
69
70
71class ModelState:
72 """Store model instance state."""
73
74 # True until the instance is first persisted (cleared after create() and by
75 # from_db; set back to True by delete()). Read by: create()/update()'s
76 # lifecycle guards, UniqueConstraint.validate (to exclude the current row
77 # from its uniqueness lookup when editing), the ModelForm created/changed
78 # message, and related-manager unsaved checks.
79 adding = True
80
81 def __init__(self) -> None:
82 self.fields_cache: dict[str, Any] = {}
83
84
85class Model(metaclass=ModelBase):
86 # Every model gets an automatic id field
87 id = types.PrimaryKeyField()
88
89 # Descriptors for other model behavior
90 query: QuerySet[Self] = QuerySet()
91 model_options: Options = Options()
92 _model_meta: Meta = Meta()
93 DoesNotExist = DoesNotExistDescriptor()
94 MultipleObjectsReturned = MultipleObjectsReturnedDescriptor()
95
96 def __init__(self, *, _from_db: bool = False, **kwargs: Any):
97 # Alias some things as locals to avoid repeat global lookups
98 cls = self.__class__
99 meta = cls._model_meta
100 _setattr = setattr
101 _DEFERRED = DEFERRED
102 _DATABASE_DEFAULT = DATABASE_DEFAULT
103
104 # Set up the storage for instance state
105 self._state = ModelState()
106
107 # Postgres owns the identity primary key -- it's generated on INSERT.
108 # Passing `id` to the constructor almost always means "give me the
109 # existing row with this id", which is a query (query.get), not a
110 # constructor argument. Reject it so a freshly constructed instance is
111 # unambiguously new. from_db() loads real rows and passes _from_db=True
112 # to skip this.
113 if not _from_db and kwargs.get("id") is not None:
114 raise ValueError(
115 f"Cannot set the auto-generated primary key 'id' when "
116 f"constructing a {cls.__name__}. To load an existing row, "
117 f"use {cls.__name__}.query.get(id=...); to create a new "
118 f"one, omit 'id' and let the database assign it."
119 )
120
121 # Process all fields from kwargs or use defaults
122 for field in meta.fields:
123 from plain.postgres.fields.related import RelatedField
124
125 is_related_object = False
126 if isinstance(field, RelatedField) and isinstance(
127 field.remote_field, ForeignObjectRel
128 ):
129 try:
130 # A foreign key is given by name -- either a related
131 # instance or a bare primary key value.
132 rel_obj = kwargs.pop(field.name)
133 is_related_object = True
134 except KeyError:
135 val = field.get_default()
136 else:
137 try:
138 val = kwargs.pop(field.name)
139 except KeyError:
140 # This is done with an exception rather than the
141 # default argument on pop because we don't want
142 # get_default() to be evaluated, and then not used.
143 # Refs #12057.
144 if field.has_db_default():
145 # DB-expression default: let Postgres evaluate it
146 # on INSERT. The compiler emits DEFAULT in the
147 # VALUES clause when it sees this sentinel.
148 val = _DATABASE_DEFAULT
149 else:
150 val = field.get_default()
151
152 if is_related_object:
153 # Assign through the descriptor so a related instance is
154 # cached and a bare key value is stored correctly.
155 if rel_obj is not _DEFERRED:
156 _setattr(self, field.name, rel_obj)
157 else:
158 if val is not _DEFERRED:
159 _setattr(self, field.name, val)
160
161 # Handle any remaining kwargs (properties or virtual fields)
162 property_names = meta._property_names
163 unexpected = ()
164 for prop, value in kwargs.items():
165 # Any remaining kwargs must correspond to properties or virtual
166 # fields.
167 if prop in property_names:
168 if value is not _DEFERRED:
169 _setattr(self, prop, value)
170 else:
171 try:
172 meta.get_field(prop)
173 except FieldDoesNotExist:
174 unexpected += (prop,)
175 else:
176 if value is not _DEFERRED:
177 _setattr(self, prop, value)
178 if unexpected:
179 unexpected_names = ", ".join(repr(n) for n in unexpected)
180 raise TypeError(
181 f"{cls.__name__}() got unexpected keyword arguments: {unexpected_names}"
182 )
183
184 super().__init__()
185
186 @classmethod
187 def from_db(cls, field_names: Iterable[str], values: Sequence[Any]) -> Model:
188 if len(values) != len(cls._model_meta.fields):
189 values_iter = iter(values)
190 values = [
191 next(values_iter) if f.name in field_names else DEFERRED
192 for f in cls._model_meta.fields
193 ]
194 # Build kwargs dict from field names and values
195 field_dict = dict(zip((f.name for f in cls._model_meta.fields), values))
196 new = cls(_from_db=True, **field_dict)
197 new._state.adding = False
198 return new
199
200 def __repr__(self) -> str:
201 return f"<{self.__class__.__name__}: {self.id}>"
202
203 def __str__(self) -> str:
204 return f"{self.__class__.__name__} object ({self.id})"
205
206 def __eq__(self, other: object) -> bool:
207 if not isinstance(other, Model):
208 return NotImplemented
209 if self.__class__ != other.__class__:
210 return False
211 my_id = self.id
212 if my_id is None:
213 return self is other
214 return my_id == other.id
215
216 def __hash__(self) -> int:
217 if self.id is None:
218 raise TypeError("Model instances without primary key value are unhashable")
219 return hash(self.id)
220
221 def __reduce__(self) -> tuple[Any, tuple[Any, ...], dict[str, Any]]:
222 data = self.__getstate__()
223 data[PLAIN_VERSION_PICKLE_KEY] = plain.runtime.__version__
224 class_id = (
225 self.model_options.package_label,
226 self.model_options.object_name,
227 )
228 return model_unpickle, (class_id,), data
229
230 def __getstate__(self) -> dict[str, Any]:
231 """Hook to allow choosing the attributes to pickle."""
232 state = self.__dict__.copy()
233 state["_state"] = copy.copy(state["_state"])
234 state["_state"].fields_cache = state["_state"].fields_cache.copy()
235 # memoryview cannot be pickled, so cast it to bytes and store
236 # separately.
237 _memoryview_attrs = []
238 for attr, value in state.items():
239 if isinstance(value, memoryview):
240 _memoryview_attrs.append((attr, bytes(value)))
241 if _memoryview_attrs:
242 state["_memoryview_attrs"] = _memoryview_attrs
243 for attr, value in _memoryview_attrs:
244 state.pop(attr)
245 return state
246
247 def __setstate__(self, state: dict[str, Any]) -> None:
248 pickled_version = state.get(PLAIN_VERSION_PICKLE_KEY)
249 if pickled_version:
250 if pickled_version != plain.runtime.__version__:
251 warnings.warn(
252 f"Pickled model instance's Plain version {pickled_version} does not "
253 f"match the current version {plain.runtime.__version__}.",
254 RuntimeWarning,
255 stacklevel=2,
256 )
257 else:
258 warnings.warn(
259 "Pickled model instance's Plain version is not specified.",
260 RuntimeWarning,
261 stacklevel=2,
262 )
263 if "_memoryview_attrs" in state:
264 for attr, value in state.pop("_memoryview_attrs"):
265 state[attr] = memoryview(value)
266 self.__dict__.update(state)
267
268 def get_deferred_fields(self) -> set[str]:
269 """
270 Return a set containing names of deferred fields for this instance.
271 """
272 return {f.name for f in self._model_meta.fields if f.name not in self.__dict__}
273
274 def refresh_from_db(self, fields: list[str] | None = None) -> None:
275 """
276 Reload field values from the database.
277
278 Fields can be used to specify which fields to reload. If fields is
279 None, then all non-deferred fields are reloaded.
280
281 When accessing deferred fields of an instance, the deferred loading
282 of the field will call this method.
283 """
284 if fields is None:
285 self._prefetched_objects_cache = {}
286 else:
287 prefetched_objects_cache = getattr(self, "_prefetched_objects_cache", {})
288 for field in fields:
289 if field in prefetched_objects_cache:
290 del prefetched_objects_cache[field]
291 fields.remove(field)
292 if not fields:
293 return
294 if any(LOOKUP_SEP in f for f in fields):
295 raise ValueError(
296 f'Found "{LOOKUP_SEP}" in fields argument. Relations and transforms '
297 "are not allowed in fields."
298 )
299
300 db_instance_qs = self._model_meta.base_queryset.filter(id=self.id)
301
302 # Use provided fields, if not set then reload all non-deferred fields.
303 deferred_fields = self.get_deferred_fields()
304 if fields is not None:
305 fields = list(fields)
306 db_instance_qs = db_instance_qs.only(*fields)
307 elif deferred_fields:
308 fields = [
309 f.name for f in self._model_meta.fields if f.name not in deferred_fields
310 ]
311 db_instance_qs = db_instance_qs.only(*fields)
312
313 db_instance = db_instance_qs.get()
314 non_loaded_fields = db_instance.get_deferred_fields()
315 for field in self._model_meta.fields:
316 if field.name in non_loaded_fields:
317 # This field wasn't refreshed - skip ahead.
318 continue
319 setattr(self, field.name, field.value_from_object(db_instance))
320 # Clear cached foreign keys.
321 if isinstance(field, RelatedField) and field.is_cached(self):
322 field.delete_cached_value(self)
323
324 # Clear cached relations.
325 for field in self._model_meta.related_objects:
326 if field.is_cached(self):
327 field.delete_cached_value(self)
328
329 def serializable_value(self, field_name: str) -> Any:
330 """
331 Return the value of the field name for this instance. If the field is
332 a foreign key, return the id value instead of the object. If there's
333 no Field object with this name on the model, return the model
334 attribute's value.
335
336 Used to serialize a field's value (in the serializer, or form output,
337 for example). Normally, you would just access the attribute directly
338 and not use this method.
339 """
340 try:
341 field = self._model_meta.get_forward_field(field_name)
342 except FieldDoesNotExist:
343 return getattr(self, field_name)
344 return field.value_from_object(self)
345
346 def create(self, *, clean_and_validate: bool = True) -> Self:
347 """INSERT this instance as a new row and return self.
348
349 Raises ValueError if the instance is already persisted (use update()).
350 With clean_and_validate (the default) the instance's shape is validated
351 first; the database enforces constraints, and a violation surfaces as
352 the ValidationError a pre-check would raise. A hand-set id is inserted
353 as given -- a collision raises IntegrityError.
354 """
355 if not self._state.adding:
356 raise ValueError(
357 f"Cannot create() a {self.__class__.__name__} that is already "
358 "persisted -- use update() instead."
359 )
360 self._prepare_related_fields_for_save(operation_name="create")
361 if clean_and_validate:
362 self.full_clean()
363 with self._mapped_write():
364 self._insert_row()
365 self._state.adding = False
366 return self
367
368 def update(
369 self,
370 *,
371 clean_and_validate: bool = True,
372 fields: Iterable[str] | None = None,
373 ) -> Self:
374 """UPDATE this instance's existing row and return self.
375
376 Raises ValueError if the instance hasn't been created yet (use
377 create()). `fields` limits the write to those columns; otherwise every
378 loaded field is written (deferred fields are skipped). Raises if no row
379 matched -- the row was deleted out from under us.
380 """
381 if self._state.adding:
382 raise ValueError(
383 f"Cannot update() a {self.__class__.__name__} that hasn't been "
384 "created yet -- use create() instead."
385 )
386 deferred_fields = self.get_deferred_fields()
387 if fields is not None:
388 if not fields:
389 return self # explicit "update nothing" -- no-op
390 fields = frozenset(fields)
391 field_names = self._model_meta._non_pk_field_names
392 non_model_fields = fields.difference(field_names)
393 if non_model_fields:
394 raise ValueError(
395 "The following fields do not exist in this model or are m2m "
396 "fields: "
397 f"{', '.join(sorted(non_model_fields))}"
398 )
399 deferred_in_update = fields & deferred_fields
400 if deferred_in_update:
401 raise FieldError(
402 "Cannot update deferred fields: "
403 f"{', '.join(sorted(deferred_in_update))}"
404 )
405 elif deferred_fields:
406 # Loaded via .only()/.defer() -- write just the loaded fields.
407 loaded = self._model_meta._non_pk_field_names - deferred_fields
408 if loaded:
409 fields = loaded
410 self._prepare_related_fields_for_save(
411 operation_name="update", field_names=fields
412 )
413 if clean_and_validate:
414 self.full_clean(exclude=deferred_fields)
415 with self._mapped_write():
416 self._update_row(fields)
417 return self
418
419 def _integrity_error_to_validation_error(
420 self, exc: psycopg.IntegrityError
421 ) -> ValidationError | None:
422 """
423 Map a Postgres constraint violation back to the constraint that raised
424 it and return the ValidationError the in-Python check produces for that
425 constraint or foreign key, or None when the violation doesn't
426 correspond to a declared constraint that can describe it (PK
427 collisions and NOT NULL — which carries no constraint name — fall
428 through to None and re-raise as the original IntegrityError). A PK
429 collision reaches here when create() inserts a hand-set id that's
430 already taken.
431 """
432 constraint_name = exc.diag.constraint_name
433 if not constraint_name:
434 return None
435 meta = self._model_meta
436 constraint = meta.constraints_by_name.get(
437 constraint_name
438 ) or meta.foreign_keys_by_constraint_name.get(constraint_name)
439 if constraint is None:
440 return None
441 error = constraint._db_violation_error(self, self.__class__)
442 if error is None:
443 return None
444 # Normalize to the same dict shape validate_constraints() produces so
445 # the error routes identically whether it was caught before or after
446 # the write: flat errors land under NON_FIELD_ERRORS, field-routed
447 # errors keep their field.
448 return ValidationError(error.update_error_dict({}))
449
450 @contextmanager
451 def _mapped_write(self) -> Iterator[None]:
452 """Run a table write inside the rollback guard, mapping a
453 declared-constraint IntegrityError to the ValidationError a pre-check
454 would have raised -- so the same violation surfaces identically whether
455 it's caught before the write or by the database. Anything we can't map
456 to a declared constraint re-raises as the original IntegrityError.
457 """
458 try:
459 with transaction.mark_for_rollback_on_error():
460 yield
461 except psycopg.IntegrityError as exc:
462 if (error := self._integrity_error_to_validation_error(exc)) is not None:
463 raise error from exc
464 raise
465
466 def _insert_row(self) -> None:
467 """INSERT this instance as a new row, filling id and any DB-default
468 fields from RETURNING. Omits id from the INSERT when unset so Postgres
469 generates the identity value."""
470 meta = self._model_meta
471 fields = list(meta.fields)
472 if self.id is None:
473 id_field = meta.get_forward_field("id")
474 fields = [f for f in fields if f is not id_field]
475 returning_fields = list(meta.db_returning_fields)
476 results = meta.base_queryset._insert(
477 [self], fields=fields, returning_fields=returning_fields or None
478 )
479 if results:
480 for value, field in zip(results[0], returning_fields):
481 setattr(self, field.name, value)
482
483 def _update_row(self, fields: Iterable[str] | None) -> None:
484 """UPDATE this instance's row from its current field values. Raise if no
485 row matched -- update() targets an existing row and has no INSERT
486 fallback (that's create())."""
487 meta = self._model_meta
488 non_pks = [f for f in meta.fields if not f.primary_key]
489 if fields:
490 non_pks = [f for f in non_pks if f.name in fields]
491 values = [(f, f.pre_save(self, False)) for f in non_pks]
492 filtered = meta.base_queryset.filter(id=self.id)
493 if not values:
494 # PK-only model -- nothing to write; the UPDATE "succeeds" as long
495 # as the row still exists. (A non-None `fields` is always validated
496 # to real non-pk columns, so it can never filter down to empty here.)
497 updated = filtered.exists()
498 else:
499 updated = filtered._update(values) > 0
500 if not updated:
501 raise psycopg.DatabaseError(
502 f"update() of {self.__class__.__name__} affected no rows -- the "
503 "row no longer exists (it may have been deleted)."
504 )
505
506 def _prepare_related_fields_for_save(
507 self, operation_name: str, field_names: Collection[str] | None = None
508 ) -> None:
509 # Ensure that a model instance without a PK hasn't been assigned to
510 # a ForeignKeyField on this model. If the field is nullable, allowing the save would result in silent data loss.
511 for field in self._model_meta.fields:
512 if field_names and field.name not in field_names:
513 continue
514 # If the related field isn't cached, then an instance hasn't been
515 # assigned and there's no need to worry about this check.
516 if isinstance(field, RelatedField) and field.is_cached(self):
517 obj = getattr(self, field.name, None)
518 if not obj:
519 continue
520 # A pk may have been assigned manually to a model instance not
521 # saved to the database, but we allow the write to proceed and
522 # rely on the database's foreign key check (mapped to a
523 # ValidationError on the field) if the row doesn't exist.
524 if obj.id is None:
525 raise ValueError(
526 f"{operation_name}() prohibited to prevent data loss due to unsaved "
527 f"related object '{field.name}'."
528 )
529 elif field.value_from_object(self) in field.empty_values:
530 # Set related object if it has been saved after an
531 # assignment.
532 setattr(self, field.name, obj)
533 # If the relationship's key was changed, clear the cached
534 # relationship. Compare the cached object's key against the raw
535 # key value -- not getattr(self, field.name), which for a
536 # foreign key returns the related object, not the key.
537 if getattr(obj, field.target_field.name) != field.value_from_object(
538 self
539 ):
540 field.delete_cached_value(self)
541
542 def delete(self) -> int:
543 """Delete this row. Returns the number of rows deleted (1 or 0).
544
545 Cascades are handled entirely by Postgres via the `on_delete`
546 clauses declared on related foreign keys.
547 """
548 if self.id is None:
549 raise ValueError(
550 f"{self.model_options.object_name} object can't be deleted because its id attribute is set "
551 "to None."
552 )
553 # Use base_queryset to bypass any user-defined filters on the public
554 # query (e.g. soft-delete scopes). An instance we have a reference to
555 # should always be deletable — custom querysets shape reads, not
556 # internal row lifecycle operations.
557 #
558 # mark_for_rollback_on_error: RESTRICT violations leave the DB
559 # transaction aborted. Mark the connection so outer atomic() blocks
560 # see the abort state even if the caller catches IntegrityError.
561 with transaction.mark_for_rollback_on_error():
562 count = self._model_meta.base_queryset.filter(id=self.id)._raw_delete()
563 id_field = self._model_meta.get_forward_field("id")
564 setattr(self, id_field.name, None)
565 # Only the id is cleared -- every other field value survives so callers
566 # can still reference a deleted row (correlate it, log it, check it's
567 # gone). The row is "new" again: create() re-inserts it (re-inserting the
568 # instance's current values, server-defaults included), update() refuses.
569 self._state.adding = True
570 return count
571
572 def get_field_display(self, field_name: str) -> str:
573 """Get the display value for a field, especially useful for fields with choices."""
574 # Get the field object from the field name
575 field = self._model_meta.get_forward_field(field_name)
576 value = field.value_from_object(self)
577
578 # If field has no choices, just return the value as string
579 if not hasattr(field, "flatchoices") or not field.flatchoices:
580 return force_str(value, strings_only=True)
581
582 # For fields with choices, look up the display value
583 choices_dict = dict(make_hashable(field.flatchoices))
584 return force_str(
585 choices_dict.get(make_hashable(value), value), strings_only=True
586 )
587
588 def _get_field_value_map(
589 self, meta: Meta | None, exclude: set[str] | None = None
590 ) -> dict[str, Value]:
591 if exclude is None:
592 exclude = set()
593 meta = meta or self._model_meta
594 return {
595 field.name: Value(field.value_from_object(self), field)
596 for field in meta.fields
597 if field.name not in exclude
598 }
599
600 def prepare_database_save(self, field: Any) -> Any:
601 if self.id is None:
602 raise ValueError(
603 f"Unsaved model instance {self!r} cannot be used in an ORM query."
604 )
605 return getattr(self, field.target_field.name)
606
607 def clean(self) -> None:
608 """
609 Hook for doing any extra model-wide validation after clean() has been
610 called on every field by self.clean_fields. Any ValidationError raised
611 by this method will not be associated with a particular field; it will
612 have a special-case association with the field defined by NON_FIELD_ERRORS.
613 """
614
615 def get_constraints(self) -> list[tuple[type[Model], list[Any]]]:
616 constraints: list[tuple[type[Model], list[Any]]] = [
617 (self.__class__, list(self.model_options.constraints))
618 ]
619 return constraints
620
621 def _unresolved_field_names(self) -> set[str]:
622 """Field names whose Python value isn't resolved yet -- either a
623 DATABASE_DEFAULT sentinel (the database fills it on INSERT) or an
624 auto_fills_on_save field (pre_save fills it just before the write).
625 Neither shape validation nor a constraint lookup can use such a value,
626 so both exclude these. Read via __dict__ to avoid triggering
627 refresh_from_db on deferred fields."""
628 names = set()
629 for f in self._model_meta.fields:
630 if self.__dict__.get(f.name) is DATABASE_DEFAULT or f.auto_fills_on_save:
631 names.add(f.name)
632 return names
633
634 def validate_constraints(self, exclude: set[str] | None = None) -> None:
635 exclude = set(exclude) if exclude else set()
636 exclude |= self._unresolved_field_names()
637 constraints = self.get_constraints()
638
639 errors: dict[str, list[ValidationError]] = {}
640 for model_class, model_constraints in constraints:
641 for constraint in model_constraints:
642 try:
643 constraint.validate(model_class, self, exclude=exclude)
644 except ValidationError as e:
645 errors = e.update_error_dict(errors)
646 if errors:
647 raise ValidationError(errors)
648
649 def full_clean(
650 self,
651 *,
652 exclude: set[str] | Iterable[str] | None = None,
653 ) -> None:
654 """
655 Validate the instance's *shape*: clean_fields() + the clean() hook.
656 Raise a ValidationError aggregating any field- or model-level errors.
657
658 Constraint checks are deliberately separate -- call
659 validate_constraints() for those. create()/update() leave them to the
660 database (mapping the resulting IntegrityError to the same
661 ValidationError a pre-check would raise); forms call
662 validate_constraints() in _post_clean to surface every violation at once.
663 """
664 errors = {}
665 exclude = set(exclude) if exclude else set()
666 exclude |= self._unresolved_field_names()
667
668 try:
669 self.clean_fields(exclude=exclude)
670 except ValidationError as e:
671 errors = e.update_error_dict(errors)
672
673 # clean() runs even if clean_fields() failed, mirroring Form.clean().
674 try:
675 self.clean()
676 except ValidationError as e:
677 errors = e.update_error_dict(errors)
678
679 if errors:
680 raise ValidationError(errors)
681
682 def clean_fields(self, exclude: set[str] | None = None) -> None:
683 """
684 Clean all fields and raise a ValidationError containing a dict
685 of all validation errors if any occur.
686 """
687 if exclude is None:
688 exclude = set()
689
690 errors = {}
691 for f in self._model_meta.fields:
692 if f.name in exclude:
693 continue
694 # Skip validation for empty fields with required=False. The developer
695 # is responsible for making sure they have a valid value.
696 raw_value = f.value_from_object(self)
697 if not f.required and raw_value in f.empty_values:
698 continue
699 try:
700 setattr(self, f.name, f.clean(raw_value, self))
701 except ValidationError as e:
702 errors[f.name] = e.error_list
703
704 if errors:
705 raise ValidationError(errors)
706
707 @classmethod
708 def preflight(cls) -> list[PreflightResult]:
709 errors: list[PreflightResult] = []
710
711 errors += [
712 *cls._check_fields(),
713 *cls._check_m2m_through_same_relationship(),
714 *cls._check_long_column_names(),
715 ]
716 clash_errors = (
717 *cls._check_id_field(),
718 *cls._check_field_name_clashes(),
719 *cls._check_model_name_db_lookup_clashes(),
720 *cls._check_property_name_related_field_accessor_clashes(),
721 *cls._check_single_primary_key(),
722 )
723 errors.extend(clash_errors)
724 # If there are field name clashes, hide consequent column name
725 # clashes.
726 if not clash_errors:
727 errors.extend(cls._check_column_name_clashes())
728 errors += [
729 *cls._check_indexes(),
730 *cls._check_ordering(),
731 *cls._check_constraints(),
732 ]
733
734 return errors
735
736 @classmethod
737 def _check_fields(cls) -> list[PreflightResult]:
738 """Perform all field checks."""
739 errors: list[PreflightResult] = []
740 for field in cls._model_meta.fields:
741 errors.extend(field.preflight(from_model=cls))
742 for field in cls._model_meta.many_to_many:
743 errors.extend(field.preflight(from_model=cls))
744 return errors
745
746 @classmethod
747 def _check_m2m_through_same_relationship(cls) -> list[PreflightResult]:
748 """Check if no relationship model is used by more than one m2m field."""
749
750 errors: list[PreflightResult] = []
751 seen_intermediary_signatures = []
752
753 fields = cls._model_meta.many_to_many
754
755 # Skip when the target or relationship model wasn't found; the field's
756 # own preflight reports those.
757 fields = (
758 f
759 for f in fields
760 if not isinstance(f.remote_field.model_ref, str)
761 and not isinstance(f.remote_field.through_ref, str)
762 )
763
764 for f in fields:
765 signature = (
766 f.remote_field.model,
767 cls,
768 f.remote_field.through,
769 f.remote_field.through_fields,
770 )
771 if signature in seen_intermediary_signatures:
772 errors.append(
773 PreflightResult(
774 fix="The model has two identical many-to-many relations "
775 f"through the intermediate model '{f.remote_field.through.model_options.label}'.",
776 obj=cls,
777 id="postgres.duplicate_many_to_many_relations",
778 )
779 )
780 else:
781 seen_intermediary_signatures.append(signature)
782 return errors
783
784 @classmethod
785 def _check_id_field(cls) -> list[PreflightResult]:
786 """Disallow user-defined fields named ``id``."""
787 if any(
788 f for f in cls._model_meta.fields if f.name == "id" and not f.auto_created
789 ):
790 return [
791 PreflightResult(
792 fix="'id' is a reserved word that cannot be used as a field name.",
793 obj=cls,
794 id="postgres.reserved_field_name_id",
795 )
796 ]
797 return []
798
799 @classmethod
800 def _check_field_name_clashes(cls) -> list[PreflightResult]:
801 """Reject fields that share a name within the same model."""
802 errors: list[PreflightResult] = []
803 used_fields = {} # name -> field
804
805 for f in cls._model_meta.fields:
806 clash = used_fields.get(f.name)
807 # Note that we may detect clash between user-defined non-unique
808 # field "id" and automatically added unique field "id", both
809 # defined at the same model. This special case is considered in
810 # _check_id_field and here we ignore it.
811 id_conflict = (
812 f.name == "id" and clash and clash.name == "id" and clash.model == cls
813 )
814 if clash and not id_conflict:
815 errors.append(
816 PreflightResult(
817 fix=f"The field '{f.name}' clashes with the field '{clash.name}' "
818 f"from model '{clash.model.model_options}'.",
819 obj=f,
820 id="postgres.field_name_clash",
821 )
822 )
823 used_fields[f.name] = f
824
825 return errors
826
827 @classmethod
828 def _check_column_name_clashes(cls) -> list[PreflightResult]:
829 # Store a list of column names which have already been used by other fields.
830 used_column_names: list[str] = []
831 errors: list[PreflightResult] = []
832
833 for f in cls._model_meta.fields:
834 column_name = f.column
835
836 # Ensure the column name is not already in use.
837 if column_name and column_name in used_column_names:
838 errors.append(
839 PreflightResult(
840 fix=f"Field '{f.name}' has column name '{column_name}' that is used by "
841 "another field.",
842 obj=cls,
843 id="postgres.db_column_clash",
844 )
845 )
846 else:
847 used_column_names.append(column_name)
848
849 return errors
850
851 @classmethod
852 def _check_model_name_db_lookup_clashes(cls) -> list[PreflightResult]:
853 errors: list[PreflightResult] = []
854 model_name = cls.__name__
855 if model_name.startswith("_") or model_name.endswith("_"):
856 errors.append(
857 PreflightResult(
858 fix=f"The model name '{model_name}' cannot start or end with an underscore "
859 "as it collides with the query lookup syntax.",
860 obj=cls,
861 id="postgres.model_name_underscore_bounds",
862 )
863 )
864 elif LOOKUP_SEP in model_name:
865 errors.append(
866 PreflightResult(
867 fix=f"The model name '{model_name}' cannot contain double underscores as "
868 "it collides with the query lookup syntax.",
869 obj=cls,
870 id="postgres.model_name_double_underscore",
871 )
872 )
873 return errors
874
875 @classmethod
876 def _check_property_name_related_field_accessor_clashes(
877 cls,
878 ) -> list[PreflightResult]:
879 errors: list[PreflightResult] = []
880 property_names = cls._model_meta._property_names
881 related_field_accessors = (
882 f.name
883 for f in cls._model_meta._get_fields(reverse=False)
884 if isinstance(f, RelatedField)
885 )
886 for accessor in related_field_accessors:
887 if accessor in property_names:
888 errors.append(
889 PreflightResult(
890 fix=f"The property '{accessor}' clashes with a related field "
891 "accessor.",
892 obj=cls,
893 id="postgres.property_related_field_clash",
894 )
895 )
896 return errors
897
898 @classmethod
899 def _check_single_primary_key(cls) -> list[PreflightResult]:
900 errors: list[PreflightResult] = []
901 if sum(1 for f in cls._model_meta.fields if f.primary_key) > 1:
902 errors.append(
903 PreflightResult(
904 fix="The model cannot have more than one field with "
905 "'primary_key=True'.",
906 obj=cls,
907 id="postgres.multiple_primary_keys",
908 )
909 )
910 return errors
911
912 @classmethod
913 def _check_indexes(cls) -> list[PreflightResult]:
914 """Check fields, names, and conditions of indexes."""
915 errors: list[PreflightResult] = []
916 references: set[str] = set()
917 for index in cls.model_options.indexes:
918 # Index name can't start with an underscore or a number
919 if index.name[0] == "_" or index.name[0].isdigit():
920 errors.append(
921 PreflightResult(
922 fix=f"The index name '{index.name}' cannot start with an underscore "
923 "or a number.",
924 obj=cls,
925 id="postgres.index_name_invalid_start",
926 ),
927 )
928 if len(index.name) > index.max_name_length:
929 errors.append(
930 PreflightResult(
931 fix="The index name '%s' cannot be longer than %d " # noqa: UP031
932 "characters." % (index.name, index.max_name_length),
933 obj=cls,
934 id="postgres.index_name_too_long",
935 ),
936 )
937 if index.contains_expressions:
938 for expression in index.expressions:
939 references.update(
940 ref[0] for ref in cls._get_expr_references(expression)
941 )
942 # Check fields referenced in indexes
943 fields = [
944 field
945 for index in cls.model_options.indexes
946 for field, _ in index.fields_orders
947 ]
948 fields += [
949 include for index in cls.model_options.indexes for include in index.include
950 ]
951 fields += references
952 errors.extend(cls._check_referenced_fields(fields, "indexes"))
953 return errors
954
955 @classmethod
956 def _check_referenced_fields(
957 cls, fields: Iterable[str], option: str
958 ) -> list[PreflightResult]:
959 # In order to avoid hitting the relation tree prematurely, we use our
960 # own fields_map instead of using get_field()
961 forward_fields_map: dict[str, Field] = {}
962 for field in cls._model_meta._get_fields(reverse=False):
963 forward_fields_map[field.name] = field
964
965 errors: list[PreflightResult] = []
966 for field_name in fields:
967 try:
968 field = forward_fields_map[field_name]
969 except KeyError:
970 errors.append(
971 PreflightResult(
972 fix=f"'{option}' refers to the nonexistent field '{field_name}'.",
973 obj=cls,
974 id="postgres.nonexistent_field_reference",
975 )
976 )
977 else:
978 from plain.postgres.fields.related import ManyToManyField
979
980 if isinstance(field, ManyToManyField):
981 errors.append(
982 PreflightResult(
983 fix=f"'{option}' refers to a ManyToManyField '{field_name}', but "
984 f"ManyToManyFields are not permitted in '{option}'.",
985 obj=cls,
986 id="postgres.m2m_field_in_meta_option",
987 )
988 )
989 return errors
990
991 @classmethod
992 def _check_ordering(cls) -> list[PreflightResult]:
993 """
994 Check "ordering" option -- is it a list of strings and do all fields
995 exist?
996 """
997
998 if not cls.model_options.ordering:
999 return []
1000
1001 if not isinstance(cls.model_options.ordering, list | tuple):
1002 return [
1003 PreflightResult(
1004 fix="'ordering' must be a tuple or list (even if you want to order by "
1005 "only one field).",
1006 obj=cls,
1007 id="postgres.ordering_not_tuple_or_list",
1008 )
1009 ]
1010
1011 errors: list[PreflightResult] = []
1012 fields = cls.model_options.ordering
1013
1014 # Skip expressions and '?' fields.
1015 fields = (f for f in fields if isinstance(f, str) and f != "?")
1016
1017 # Convert "-field" to "field".
1018 fields = (f.removeprefix("-") for f in fields)
1019
1020 # Separate related fields and non-related fields.
1021 _fields = []
1022 related_fields = []
1023 for f in fields:
1024 if LOOKUP_SEP in f:
1025 related_fields.append(f)
1026 else:
1027 _fields.append(f)
1028 fields = _fields
1029
1030 # Check related fields.
1031 for field in related_fields:
1032 _cls = cls
1033 fld = None
1034 for part in field.split(LOOKUP_SEP):
1035 try:
1036 if _cls is None:
1037 # The previous part was not a relation, so there is
1038 # no model left to look the next part up on.
1039 raise FieldDoesNotExist(part)
1040 fld = _cls._model_meta.get_field(part)
1041 if isinstance(fld, RelatedField):
1042 _cls = fld.path_infos[-1].to_meta.model
1043 else:
1044 _cls = None
1045 except (FieldDoesNotExist, AttributeError):
1046 if fld is None or (
1047 not isinstance(fld, Field)
1048 or (
1049 fld.get_transform(part) is None
1050 and fld.get_lookup(part) is None
1051 )
1052 ):
1053 errors.append(
1054 PreflightResult(
1055 fix="'ordering' refers to the nonexistent field, "
1056 f"related field, or lookup '{field}'.",
1057 obj=cls,
1058 id="postgres.ordering_nonexistent_field",
1059 )
1060 )
1061
1062 # Check for invalid or nonexistent fields in ordering.
1063 invalid_fields = []
1064
1065 # Any field name that is not present in field_names does not exist.
1066 # Also, ordering by m2m fields is not allowed.
1067 meta = cls._model_meta
1068 valid_fields = set(
1069 chain.from_iterable(
1070 (f.field.related_query_name(),)
1071 if isinstance(f, ForeignObjectRel)
1072 else (f.name,)
1073 for f in chain(meta.fields, meta.related_objects)
1074 )
1075 )
1076
1077 invalid_fields.extend(set(fields) - valid_fields)
1078
1079 for invalid_field in invalid_fields:
1080 errors.append(
1081 PreflightResult(
1082 fix="'ordering' refers to the nonexistent field, related "
1083 f"field, or lookup '{invalid_field}'.",
1084 obj=cls,
1085 id="postgres.ordering_nonexistent_field",
1086 )
1087 )
1088 return errors
1089
1090 @classmethod
1091 def _check_long_column_names(cls) -> list[PreflightResult]:
1092 """
1093 Check that any auto-generated column names are shorter than the limits
1094 for each database in which the model will be created.
1095 """
1096 errors: list[PreflightResult] = []
1097
1098 # PostgreSQL has a 63-character limit on identifier names and doesn't
1099 # silently truncate, so we check for names that are too long
1100 allowed_len = MAX_NAME_LENGTH
1101
1102 for f in cls._model_meta.fields:
1103 column_name = f.column
1104
1105 # Check if column name is too long for the database.
1106 if column_name is not None and len(column_name) > allowed_len:
1107 errors.append(
1108 PreflightResult(
1109 fix=f'Column name too long for field "{column_name}". '
1110 f'Maximum length is "{allowed_len}" for the database.',
1111 obj=cls,
1112 id="postgres.column_name_too_long",
1113 )
1114 )
1115
1116 for f in cls._model_meta.many_to_many:
1117 # Skip nonexistent models.
1118 if isinstance(f.remote_field.through_ref, str):
1119 continue
1120
1121 # Check if column name for the M2M field is too long for the database.
1122 for m2m in f.remote_field.through_ref._model_meta.fields:
1123 rel_name = m2m.column
1124 if rel_name is not None and len(rel_name) > allowed_len:
1125 errors.append(
1126 PreflightResult(
1127 fix="Column name too long for M2M field "
1128 f'"{rel_name}". Maximum length is "{allowed_len}" for the database.',
1129 obj=cls,
1130 id="postgres.m2m_column_name_too_long",
1131 )
1132 )
1133
1134 return errors
1135
1136 @classmethod
1137 def _get_expr_references(cls, expr: Any) -> Iterator[tuple[str, ...]]:
1138 if isinstance(expr, Q):
1139 for child in expr.children:
1140 if isinstance(child, tuple):
1141 lookup, value = child
1142 yield tuple(lookup.split(LOOKUP_SEP))
1143 yield from cls._get_expr_references(value)
1144 else:
1145 yield from cls._get_expr_references(child)
1146 elif isinstance(expr, F):
1147 yield tuple(expr.name.split(LOOKUP_SEP))
1148 elif hasattr(expr, "get_source_expressions"):
1149 for src_expr in expr.get_source_expressions():
1150 yield from cls._get_expr_references(src_expr)
1151
1152 @classmethod
1153 def _check_constraints(cls) -> list[PreflightResult]:
1154 errors: list[PreflightResult] = []
1155 fields = set(
1156 chain.from_iterable(
1157 (*constraint.fields, *constraint.include)
1158 for constraint in cls.model_options.constraints
1159 if isinstance(constraint, UniqueConstraint)
1160 )
1161 )
1162 references = set()
1163 for constraint in cls.model_options.constraints:
1164 if isinstance(constraint, UniqueConstraint):
1165 if isinstance(constraint.condition, Q):
1166 references.update(cls._get_expr_references(constraint.condition))
1167 if constraint.contains_expressions:
1168 for expression in constraint.expressions:
1169 references.update(cls._get_expr_references(expression))
1170 elif isinstance(constraint, CheckConstraint):
1171 if isinstance(constraint.check, Q):
1172 references.update(cls._get_expr_references(constraint.check))
1173 if any(isinstance(expr, RawSQL) for expr in constraint.check.flatten()):
1174 errors.append(
1175 PreflightResult(
1176 fix=f"Check constraint {constraint.name!r} contains "
1177 f"RawSQL() expression and won't be validated "
1178 f"during the model full_clean(). "
1179 "Silence this warning if you don't care about it.",
1180 warning=True,
1181 obj=cls,
1182 id="postgres.constraint_name_collision_autogenerated",
1183 ),
1184 )
1185 for field_name, *lookups in references:
1186 fields.add(field_name)
1187 if not lookups:
1188 # If it has no lookups it cannot result in a JOIN.
1189 continue
1190 try:
1191 field = cls._model_meta.get_field(field_name)
1192 from plain.postgres.fields.related import ManyToManyField
1193 from plain.postgres.fields.reverse_related import ForeignKeyRel
1194
1195 if not isinstance(field, RelatedField) or isinstance(
1196 field, (ManyToManyField, ForeignKeyRel)
1197 ):
1198 continue
1199 except FieldDoesNotExist:
1200 continue
1201 # JOIN must happen at the first lookup.
1202 first_lookup = lookups[0]
1203 if (
1204 hasattr(field, "get_transform")
1205 and hasattr(field, "get_lookup")
1206 and field.get_transform(first_lookup) is None
1207 and field.get_lookup(first_lookup) is None
1208 ):
1209 errors.append(
1210 PreflightResult(
1211 fix=f"'constraints' refers to the joined field '{LOOKUP_SEP.join([field_name] + lookups)}'.",
1212 obj=cls,
1213 id="postgres.constraint_refers_to_joined_field",
1214 )
1215 )
1216 errors.extend(cls._check_referenced_fields(fields, "constraints"))
1217 return errors
1218
1219
1220########
1221# MISC #
1222########
1223
1224
1225def model_unpickle(model_id: tuple[str, str] | type[Model]) -> Model:
1226 """Used to unpickle Model subclasses with deferred fields."""
1227 if isinstance(model_id, tuple):
1228 model = models_registry.get_model(*model_id)
1229 else:
1230 # Backwards compat - the model was cached directly in earlier versions.
1231 model = model_id
1232 return model.__new__(model)
1233
1234
1235# Pickle protocol marker - functions don't normally have this attribute
1236model_unpickle.__safe_for_unpickle__ = True # ty: ignore[unresolved-attribute]