1from __future__ import annotations
2
3import copy
4from collections.abc import Callable, Sequence
5from functools import cached_property, partial
6from typing import TYPE_CHECKING, Any, Self, cast
7
8from plain.exceptions import ValidationError
9from plain.postgres.constants import LOOKUP_SEP
10from plain.postgres.deletion import SET_NULL, OnDelete
11from plain.postgres.exceptions import FieldDoesNotExist
12from plain.postgres.query_utils import PathInfo
13from plain.postgres.utils import generate_fk_constraint_name, make_model_tuple
14from plain.preflight import PreflightResult
15
16from ..registry import models_registry
17from . import Field
18from .base import ColumnField
19from .mixins import FieldCacheMixin
20from .related_descriptors import (
21 ForwardForeignKeyDescriptor,
22 ForwardManyToManyDescriptor,
23)
24from .related_lookups import (
25 RelatedExact,
26 RelatedGreaterThan,
27 RelatedGreaterThanOrEqual,
28 RelatedIn,
29 RelatedIsNull,
30 RelatedLessThan,
31 RelatedLessThanOrEqual,
32)
33from .reverse_related import ForeignKeyRel, ManyToManyRel
34
35if TYPE_CHECKING:
36 from plain.postgres.base import Model
37 from plain.postgres.connection import DatabaseConnection
38 from plain.postgres.fields.reverse_related import ForeignObjectRel
39
40RECURSIVE_RELATIONSHIP_CONSTANT = "self"
41
42
43def resolve_relation(
44 scope_model: type[Model], relation: type[Model] | str
45) -> type[Model] | str:
46 """
47 Transform relation into a model or fully-qualified model string of the form
48 "package_label.ModelName", relative to scope_model.
49
50 The relation argument can be:
51 * RECURSIVE_RELATIONSHIP_CONSTANT, i.e. the string "self", in which case
52 the model argument will be returned.
53 * A bare model name without an package_label, in which case scope_model's
54 package_label will be prepended.
55 * An "package_label.ModelName" string.
56 * A model class, which will be returned unchanged.
57 """
58 # Check for recursive relations
59 if relation == RECURSIVE_RELATIONSHIP_CONSTANT:
60 relation = scope_model
61
62 # Look for an "app.Model" relation
63 if isinstance(relation, str) and "." not in relation:
64 relation = f"{scope_model.model_options.package_label}.{relation}"
65
66 return relation
67
68
69def lazy_related_operation(
70 function: Any, model: type[Model], *related_models: type[Model] | str, **kwargs: Any
71) -> None:
72 """
73 Schedule `function` to be called once `model` and all `related_models`
74 have been imported and registered with the app registry. `function` will
75 be called with the newly-loaded model classes as its positional arguments,
76 plus any optional keyword arguments.
77
78 The `model` argument must be a model class. Each subsequent positional
79 argument is another model, or a reference to another model - see
80 `resolve_relation()` for the various forms these may take. Any relative
81 references will be resolved relative to `model`.
82
83 This is a convenience wrapper for `Packages.lazy_model_operation` - the app
84 registry model used is the one found in `model._model_meta.models_registry`.
85 """
86 models = [model] + [resolve_relation(model, rel) for rel in related_models]
87 model_keys = (make_model_tuple(m) for m in models)
88 models_registry = model._model_meta.models_registry
89 return models_registry.lazy_model_operation(
90 partial(function, **kwargs), *model_keys
91 )
92
93
94class RelatedField(FieldCacheMixin, Field):
95 """Base class that all relational fields inherit from."""
96
97 non_migration_attrs = (
98 *Field.non_migration_attrs,
99 "related_query_name",
100 )
101
102 # RelatedField always has a remote_field (never None)
103 remote_field: ForeignObjectRel
104 # path_infos and reverse_path_infos are implemented as @cached_property
105 # in subclasses (ForeignKeyField, ManyToManyField)
106 path_infos: list[PathInfo]
107 reverse_path_infos: list[PathInfo]
108 # Set by ForeignKeyField / ManyToManyField in their __init__; declared
109 # here so RelatedField methods (deconstruct, related_query_name) can
110 # reference them without isinstance-narrowing.
111 _related_query_name: str | None
112
113 # No __init__: ForeignKeyField and ManyToManyField each set
114 # _related_query_name and remote_field themselves.
115
116 def __deepcopy__(self, memodict: dict[int, Any]) -> Self:
117 # Handle remote_field deepcopy for RelatedFields
118 obj = super().__deepcopy__(memodict)
119 obj.remote_field = copy.copy(self.remote_field)
120 if hasattr(self.remote_field, "field") and self.remote_field.field is self:
121 obj.remote_field.field = obj
122 return obj
123
124 @cached_property
125 def related_model(self) -> type[Model]:
126 # Can't cache this property until all the models are loaded.
127 models_registry.check_ready()
128 return self.remote_field.model
129
130 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
131 return [
132 *super().preflight(**kwargs),
133 *self._check_related_query_name_is_valid(),
134 *self._check_relation_model_exists(),
135 *self._check_clashes(),
136 ]
137
138 def _check_related_query_name_is_valid(self) -> list[PreflightResult]:
139 # Always validate related_query_name since it's still used for ORM queries
140 # (e.g., User.query.filter(articles__title="..."))
141 rel_query_name = self.related_query_name()
142 errors: list[PreflightResult] = []
143 if rel_query_name.endswith("_"):
144 errors.append(
145 PreflightResult(
146 fix=(
147 f"Reverse query name '{rel_query_name}' must not end with an underscore. "
148 "Use a different related_query_name."
149 ),
150 obj=self,
151 id="fields.related_field_accessor_clash",
152 )
153 )
154 if LOOKUP_SEP in rel_query_name:
155 errors.append(
156 PreflightResult(
157 fix=(
158 f"Reverse query name '{rel_query_name}' must not contain '{LOOKUP_SEP}'. "
159 "Use a different related_query_name."
160 ),
161 obj=self,
162 id="fields.related_field_query_name_clash",
163 )
164 )
165 return errors
166
167 def _check_relation_model_exists(self) -> list[PreflightResult]:
168 rel_is_missing = (
169 self.remote_field.model_ref not in self.meta.models_registry.get_models()
170 )
171 rel_is_string = isinstance(self.remote_field.model_ref, str)
172 model_name = (
173 self.remote_field.model_ref
174 if rel_is_string
175 else self.remote_field.model_ref.model_options.object_name
176 )
177 if rel_is_missing and rel_is_string:
178 return [
179 PreflightResult(
180 fix=(
181 f"Field defines a relation with model '{model_name}', which is not "
182 "installed. Ensure the model's package is registered."
183 ),
184 obj=self,
185 id="fields.related_model_not_installed",
186 )
187 ]
188 return []
189
190 def _check_clashes(self) -> list[PreflightResult]:
191 """Check accessor and reverse query name clashes."""
192 errors: list[PreflightResult] = []
193
194 # Skip if the target model is still an unresolved string reference;
195 # _check_relation_model_exists reports that case.
196 if isinstance(self.remote_field.model_ref, str):
197 return []
198
199 # Consider that we are checking field `Model.foreign` and the models
200 # are:
201 #
202 # class Target(models.Model):
203 # model = models.IntegerField()
204 # model_set = models.IntegerField()
205 #
206 # class Model(models.Model):
207 # foreign = models.ForeignKeyField(Target)
208 # m2m = models.ManyToManyField(Target)
209
210 # rel_options.object_name == "Target"
211 rel_meta = self.remote_field.model._model_meta
212 rel_options = self.remote_field.model.model_options
213 rel_query_name = self.related_query_name() # i. e. "model"
214 # i.e. "package_label.Model.field".
215 field_name = f"{self.model.model_options.label}.{self.name}"
216
217 # Check clashes between reverse query name of `field`
218 # and any other field name.
219 potential_clashes = rel_meta.fields + rel_meta.many_to_many
220 for clash_field in potential_clashes:
221 # i.e. "package_label.Target.model_set".
222 clash_name = f"{rel_options.label}.{clash_field.name}"
223 if clash_field.name == rel_query_name:
224 errors.append(
225 PreflightResult(
226 fix=(
227 f"Reverse query name for '{field_name}' clashes with field name '{clash_name}'. "
228 f"Rename field '{clash_name}' or use a different related_query_name."
229 ),
230 obj=self,
231 id="fields.related_accessor_clash_manager",
232 )
233 )
234
235 return errors
236
237 def db_type(self) -> str | None:
238 # By default related field will not have a column as it relates to
239 # columns from another table.
240 return None
241
242 def unqualified_db_type(self) -> str | None:
243 return self.rel_db_type()
244
245 def contribute_to_class(self, cls: type[Model], name: str) -> None:
246 super().contribute_to_class(cls, name)
247
248 self.meta = cls._model_meta
249
250 if self.remote_field.related_query_name:
251 related_query_name = self.remote_field.related_query_name % {
252 "class": cls.__name__.lower(),
253 "package_label": cls.model_options.package_label.lower(),
254 }
255 self.remote_field.related_query_name = related_query_name
256
257 def resolve_related_class(
258 model: type[Model], related: type[Model], field: RelatedField
259 ) -> None:
260 field.remote_field.model_ref = related
261 field.do_related_class(related, model)
262
263 lazy_related_operation(
264 resolve_related_class,
265 cls,
266 self.remote_field.model_ref,
267 field=self,
268 )
269
270 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
271 name, path, args, kwargs = super().deconstruct()
272 if self._related_query_name is not None:
273 kwargs["related_query_name"] = self._related_query_name
274 return name, path, args, kwargs
275
276 def set_attributes_from_rel(self) -> None:
277 self.name = self.name or (
278 self.remote_field.model.model_options.model_name + "_" + "id"
279 )
280
281 def do_related_class(self, other: type[Model], cls: type[Model]) -> None:
282 self.set_attributes_from_rel()
283
284 def related_query_name(self) -> str:
285 """
286 Define the name that can be used to identify this related object in a
287 table-spanning query.
288 """
289 return (
290 self.remote_field.related_query_name or self.model.model_options.model_name
291 )
292
293 @property
294 def target_field(self) -> Field:
295 """
296 The remote field this relation filters against. Read dynamically by the
297 related-lookup prep path (`output_field.target_field`). ForeignKeyField
298 overrides this with a direct, cached lookup; ManyToManyField uses this
299 path-based form.
300 """
301 return self.path_infos[-1].target_field
302
303 def get_cache_name(self) -> str:
304 return self.name
305
306
307class ForeignKeyField(ColumnField, RelatedField):
308 # Narrow the base class's `ForeignObjectRel` annotation — a FK's remote_field
309 # is always a ForeignKeyRel with a concrete on_delete action.
310 remote_field: ForeignKeyRel
311
312 """
313 Provide a many-to-one relation by adding a column to the local model
314 to hold the remote value.
315
316 ForeignKeyField targets the primary key (id) of the remote model.
317 """
318
319 non_migration_attrs = (
320 *RelatedField.non_migration_attrs,
321 *ColumnField.non_migration_attrs,
322 "on_delete",
323 )
324
325 empty_strings_allowed = False
326
327 does_not_exist_error_message = "%(object_name)s with id %(value)s does not exist."
328
329 def __init__(
330 self,
331 to: type[Model] | str,
332 on_delete: OnDelete,
333 related_query_name: str | None = None,
334 *,
335 required: bool = True,
336 allow_null: bool = False,
337 validators: Sequence[Callable[..., Any]] = (),
338 ):
339 # `default` and `choices` are intentionally not accepted: a hardcoded
340 # FK id default is a portability/existence footgun, and the related
341 # model itself already defines the valid set.
342 if not isinstance(to, str):
343 try:
344 to.model_options.model_name # noqa: B018 — duck-type probe
345 except AttributeError:
346 raise TypeError(
347 f"{self.__class__.__name__}({to!r}) is invalid. First parameter to ForeignKeyField must be "
348 f"either a model, a model name, or the string {RECURSIVE_RELATIONSHIP_CONSTANT!r}"
349 )
350 if not isinstance(on_delete, OnDelete):
351 raise TypeError(
352 "on_delete must be one of plain.postgres.CASCADE, SET_NULL, "
353 f"or RESTRICT; got {on_delete!r}"
354 )
355
356 super().__init__(
357 required=required,
358 allow_null=allow_null,
359 validators=validators,
360 )
361 self._related_query_name = related_query_name
362 self.remote_field = ForeignKeyRel(
363 field=self,
364 to=to,
365 on_delete=on_delete,
366 related_query_name=related_query_name,
367 )
368
369 def __copy__(self) -> ForeignKeyField:
370 obj = super().__copy__()
371 # Remove cached values that depend on the (possibly not-yet-resolved)
372 # remote model.
373 obj.__dict__.pop("target_field", None)
374 obj.__dict__.pop("path_infos", None)
375 obj.__dict__.pop("reverse_path_infos", None)
376 return obj
377
378 def get_local_related_value(self, instance: Model) -> Any:
379 # The local field is the foreign key itself; read its raw key value, not
380 # the related object the descriptor returns.
381 return self._get_raw_value(instance)
382
383 def get_foreign_related_value(self, instance: Model) -> Any:
384 # A foreign key always points at the remote instance's id.
385 return instance.id
386
387 def get_joining_columns(self, reverse_join: bool = False) -> tuple[str, str]:
388 # A foreign key joins one column pair: this field's column and the
389 # target's id column.
390 if reverse_join:
391 return (self.target_field.column, self.column)
392 return (self.column, self.target_field.column)
393
394 @cached_property
395 def path_infos(self) -> list[PathInfo]:
396 """Get path from this field to the related model."""
397 meta = self.remote_field.model._model_meta
398 from_meta = self.model._model_meta
399 return [
400 PathInfo(
401 from_meta=from_meta,
402 to_meta=meta,
403 target_field=self.target_field,
404 join_field=self,
405 m2m=False,
406 direct=True,
407 )
408 ]
409
410 @cached_property
411 def reverse_path_infos(self) -> list[PathInfo]:
412 """Get path from the related model to this field's model."""
413 meta = self.model._model_meta
414 from_meta = self.remote_field.model._model_meta
415 return [
416 PathInfo(
417 from_meta=from_meta,
418 to_meta=meta,
419 target_field=meta.get_forward_field("id"),
420 join_field=self.remote_field,
421 # The reverse of a foreign key always fans out to many rows.
422 m2m=True,
423 direct=False,
424 )
425 ]
426
427 def contribute_to_class(self, cls: type[Model], name: str) -> None:
428 super().contribute_to_class(cls, name)
429 setattr(cls, name, ForwardForeignKeyDescriptor(self))
430
431 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
432 return [
433 *super().preflight(**kwargs),
434 *self._check_on_delete(),
435 ]
436
437 def _check_on_delete(self) -> list[PreflightResult]:
438 on_delete = getattr(self.remote_field, "on_delete", None)
439 results: list[PreflightResult] = []
440 if on_delete is SET_NULL and not self.allow_null:
441 results.append(
442 PreflightResult(
443 fix=(
444 "Field specifies on_delete=SET_NULL, but cannot be null. "
445 "Set allow_null=True argument on the field, or change the on_delete rule."
446 ),
447 obj=self,
448 id="fields.foreign_key_null_constraint_violation",
449 )
450 )
451 return results
452
453 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
454 name, path, args, kwargs = super().deconstruct()
455 kwargs["on_delete"] = self.remote_field.on_delete
456
457 if isinstance(self.remote_field.model_ref, str):
458 if "." in self.remote_field.model_ref:
459 package_label, model_name = self.remote_field.model_ref.split(".")
460 kwargs["to"] = f"{package_label}.{model_name.lower()}"
461 else:
462 kwargs["to"] = self.remote_field.model_ref.lower()
463 else:
464 kwargs["to"] = self.remote_field.model_ref.model_options.label_lower
465
466 return name, path, args, kwargs
467
468 def to_python(self, value: Any) -> Any:
469 return self.target_field.to_python(value)
470
471 @cached_property
472 def target_field(self) -> Field:
473 """A foreign key points at exactly one column: the remote model's id."""
474 return self.remote_field.model._model_meta.get_forward_field("id")
475
476 def set_attributes_from_name(self, name: str) -> None:
477 super().set_attributes_from_name(name)
478 # The raw key value lives in instance.__dict__ under the field name
479 # ("author"), reached only through the ForwardForeignKeyDescriptor --
480 # there is no separate "author_id" attribute. The database column
481 # keeps the historical _id suffix.
482 self.column = f"{self.name}_id"
483
484 def _get_raw_value(self, instance: Model) -> Any:
485 """Return the raw related key stored on the instance.
486
487 Reads instance.__dict__ directly -- never through the descriptor, which
488 would yield the related object. If the foreign key column was deferred
489 (.only()/.defer()) it is loaded first, so save/serialize/validate paths
490 never mistake a deferred column for a NULL value.
491 """
492 if self.name not in instance.__dict__:
493 instance.refresh_from_db(fields=[self.name])
494 return instance.__dict__.get(self.name)
495
496 def pre_save(self, model_instance: Model, add: bool) -> Any:
497 # Return the raw related key for INSERT/UPDATE, not the related object
498 # that attribute access yields.
499 return self._get_raw_value(model_instance)
500
501 def value_from_object(self, obj: Model) -> Any:
502 return self._get_raw_value(obj)
503
504 def db_constraint_name(self) -> str:
505 """The name Postgres reports in ``err.diag.constraint_name`` when this
506 foreign key is violated."""
507 assert self.model is not None
508 return generate_fk_constraint_name(
509 self.model.model_options.db_table,
510 self.column,
511 self.target_field.model.model_options.db_table,
512 self.target_field.column,
513 )
514
515 def _db_violation_error(
516 self, instance: Model, model: type[Model]
517 ) -> ValidationError:
518 """The ValidationError for a ForeignKeyViolation on a write of
519 ``instance`` — the referenced row doesn't exist. Routed to this field,
520 like a single-field unique violation.
521
522 Same signature as ``BaseConstraint._db_violation_error`` so the write
523 path treats a foreign key and a declared constraint alike; ``model``
524 is unused here."""
525 assert self.name is not None
526 error = ValidationError(
527 self.does_not_exist_error_message,
528 code="invalid_choice",
529 params={
530 "object_name": self.related_model.model_options.object_name,
531 "value": self.value_from_object(instance),
532 },
533 )
534 return ValidationError({self.name: [error]})
535
536 def get_db_prep_save(self, value: Any, connection: DatabaseConnection) -> Any:
537 if value is None or (
538 value == "" and not self.target_field.empty_strings_allowed
539 ):
540 return None
541 else:
542 return self.target_field.get_db_prep_save(value, connection=connection)
543
544 def get_db_prep_value(
545 self, value: Any, connection: DatabaseConnection, prepared: bool = False
546 ) -> Any:
547 return self.target_field.get_db_prep_value(value, connection, prepared)
548
549 def get_prep_value(self, value: Any) -> Any:
550 return self.target_field.get_prep_value(value)
551
552 def db_type(self) -> str | None:
553 return self.target_field.rel_db_type()
554
555 def cast_db_type(self) -> str | None:
556 return self.target_field.cast_db_type()
557
558 def get_col(self, alias: str | None, output_field: Field | None = None) -> Any:
559 if output_field is None:
560 # A foreign key resolves to its target's id column.
561 output_field = self.target_field
562 return super().get_col(alias, output_field)
563
564
565# Register lookups for ForeignKey
566ForeignKeyField.register_lookup(RelatedIn)
567ForeignKeyField.register_lookup(RelatedExact)
568ForeignKeyField.register_lookup(RelatedLessThan)
569ForeignKeyField.register_lookup(RelatedGreaterThan)
570ForeignKeyField.register_lookup(RelatedGreaterThanOrEqual)
571ForeignKeyField.register_lookup(RelatedLessThanOrEqual)
572ForeignKeyField.register_lookup(RelatedIsNull)
573
574
575class ManyToManyField(RelatedField):
576 """
577 Provide a many-to-many relation by using an intermediary model that
578 holds two ForeignKeyField fields pointed at the two sides of the relation.
579
580 Unless a ``through`` model was provided, ManyToManyField will use the
581 create_many_to_many_intermediary_model factory to automatically generate
582 the intermediary model.
583 """
584
585 # ManyToManyField uses ManyToManyRel which has through/through_fields
586 remote_field: ManyToManyRel
587
588 def __init__(
589 self,
590 to: type[Model] | str,
591 *,
592 through: type[Model] | str,
593 through_fields: tuple[str, str] | None = None,
594 related_query_name: str | None = None,
595 symmetrical: bool | None = None,
596 ):
597 # M2M has no database column, so `required`, `allow_null`, `default`,
598 # `validators`, and `choices` are intentionally not accepted. Membership
599 # is managed through the related manager.
600 if not isinstance(to, str):
601 try:
602 to._model_meta # noqa: B018 — duck-type probe
603 except AttributeError:
604 raise TypeError(
605 f"{self.__class__.__name__}({to!r}) is invalid. First parameter to ManyToManyField "
606 f"must be either a model, a model name, or the string {RECURSIVE_RELATIONSHIP_CONSTANT!r}"
607 )
608
609 if symmetrical is None:
610 symmetrical = to == RECURSIVE_RELATIONSHIP_CONSTANT
611
612 if not through:
613 raise ValueError("ManyToManyField must have a 'through' argument.")
614
615 self.remote_field = ManyToManyRel(
616 field=self,
617 to=to,
618 related_query_name=related_query_name,
619 symmetrical=symmetrical,
620 through=through,
621 through_fields=through_fields,
622 )
623
624 super().__init__()
625 self._related_query_name = related_query_name
626
627 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
628 return [
629 *super().preflight(**kwargs),
630 *self._check_relationship_model(**kwargs),
631 *self._check_table_uniqueness(**kwargs),
632 ]
633
634 def _check_relationship_model(
635 self, from_model: type[Model] | None = None, **kwargs: Any
636 ) -> list[PreflightResult]:
637 through_ref = self.remote_field.through_ref
638 if isinstance(through_ref, str):
639 qualified_model_name = through_ref
640 else:
641 qualified_model_name = (
642 f"{through_ref.model_options.package_label}.{through_ref.__name__}"
643 )
644
645 errors = []
646
647 if through_ref not in self.meta.models_registry.get_models():
648 # The relationship model is not installed.
649 errors.append(
650 PreflightResult(
651 fix=(
652 "Field specifies a many-to-many relation through model "
653 f"'{qualified_model_name}', which has not been installed. "
654 "Ensure the through model is properly defined and installed."
655 ),
656 obj=self,
657 id="fields.m2m_through_model_not_installed",
658 )
659 )
660
661 else:
662 assert from_model is not None, (
663 "ManyToManyField with intermediate "
664 "tables cannot be checked if you don't pass the model "
665 "where the field is attached to."
666 )
667 # Set some useful local variables
668 to_model = resolve_relation(from_model, self.remote_field.model_ref)
669 from_model_name = from_model.model_options.object_name
670 if isinstance(to_model, str):
671 to_model_name = to_model
672 else:
673 to_model_name = to_model.model_options.object_name
674 relationship_model_name = (
675 self.remote_field.through.model_options.object_name
676 )
677 self_referential = from_model == to_model
678 # Count foreign keys in intermediate model
679 if self_referential:
680 seen_self = sum(
681 from_model == field.remote_field.model_ref
682 for field in self.remote_field.through._model_meta.fields
683 if isinstance(field, RelatedField)
684 )
685
686 if seen_self > 2 and not self.remote_field.through_fields:
687 errors.append(
688 PreflightResult(
689 fix=(
690 "The model is used as an intermediate model by "
691 f"'{self}', but it has more than two foreign keys "
692 f"to '{from_model_name}', which is ambiguous. "
693 "Use through_fields to specify which two foreign keys "
694 "Plain should use."
695 ),
696 obj=self.remote_field.through,
697 id="fields.m2m_through_model_ambiguous_fks",
698 )
699 )
700
701 else:
702 # Count foreign keys in relationship model
703 seen_from = sum(
704 from_model == field.remote_field.model_ref
705 for field in self.remote_field.through._model_meta.fields
706 if isinstance(field, RelatedField)
707 )
708 seen_to = sum(
709 to_model == field.remote_field.model_ref
710 for field in self.remote_field.through._model_meta.fields
711 if isinstance(field, RelatedField)
712 )
713
714 if seen_from > 1 and not self.remote_field.through_fields:
715 errors.append(
716 PreflightResult(
717 fix=(
718 "The model is used as an intermediate model by "
719 f"'{self}', but it has more than one foreign key "
720 f"from '{from_model_name}', which is ambiguous. You must specify "
721 "which foreign key Plain should use via the "
722 "through_fields keyword argument. "
723 "If you want to create a recursive relationship, "
724 f'use ManyToManyField("{RECURSIVE_RELATIONSHIP_CONSTANT}", through="{relationship_model_name}").'
725 ),
726 obj=self,
727 id="fields.m2m_through_model_invalid_recursive_from",
728 )
729 )
730
731 if seen_to > 1 and not self.remote_field.through_fields:
732 errors.append(
733 PreflightResult(
734 fix=(
735 "The model is used as an intermediate model by "
736 f"'{self}', but it has more than one foreign key "
737 f"to '{to_model_name}', which is ambiguous. You must specify "
738 "which foreign key Plain should use via the "
739 "through_fields keyword argument. "
740 "If you want to create a recursive relationship, "
741 f'use ManyToManyField("{RECURSIVE_RELATIONSHIP_CONSTANT}", through="{relationship_model_name}").'
742 ),
743 obj=self,
744 id="fields.m2m_through_model_invalid_recursive_to",
745 )
746 )
747
748 if seen_from == 0 or seen_to == 0:
749 errors.append(
750 PreflightResult(
751 fix=(
752 "The model is used as an intermediate model by "
753 f"'{self}', but it does not have a foreign key to '{from_model_name}' or '{to_model_name}'. "
754 "Add the required foreign keys to the through model."
755 ),
756 obj=self.remote_field.through,
757 id="fields.m2m_through_model_missing_fk",
758 )
759 )
760
761 # Validate `through_fields`.
762 if self.remote_field.through_fields is not None:
763 # Validate that we're given an iterable of at least two items
764 # and that none of them is "falsy".
765 if not (
766 len(self.remote_field.through_fields) >= 2
767 and self.remote_field.through_fields[0]
768 and self.remote_field.through_fields[1]
769 ):
770 errors.append(
771 PreflightResult(
772 fix=(
773 "Field specifies 'through_fields' but does not provide "
774 "the names of the two link fields that should be used "
775 f"for the relation through model '{qualified_model_name}'. "
776 "Make sure you specify 'through_fields' as "
777 "through_fields=('field1', 'field2')."
778 ),
779 obj=self,
780 id="fields.m2m_through_fields_wrong_length",
781 )
782 )
783
784 # Validate the given through fields -- they should be actual
785 # fields on the through model, and also be foreign keys to the
786 # expected models.
787 else:
788 assert from_model is not None, (
789 "ManyToManyField with intermediate "
790 "tables cannot be checked if you don't pass the model "
791 "where the field is attached to."
792 )
793
794 source, through, target = (
795 from_model,
796 self.remote_field.through,
797 self.remote_field.model_ref,
798 )
799 source_field_name, target_field_name = self.remote_field.through_fields[
800 :2
801 ]
802
803 for field_name, related_model in (
804 (source_field_name, source),
805 (target_field_name, target),
806 ):
807 # The target may still be an unresolved "package.Model" string.
808 related_model_name = (
809 related_model
810 if isinstance(related_model, str)
811 else related_model.model_options.object_name
812 )
813 possible_field_names: list[str] = []
814 for f in through._model_meta.fields:
815 if (
816 isinstance(f, RelatedField)
817 and f.remote_field.model_ref == related_model
818 ):
819 possible_field_names.append(f.name)
820 if possible_field_names:
821 fix = (
822 "Did you mean one of the following foreign keys to '{}': "
823 "{}?".format(
824 related_model_name,
825 ", ".join(possible_field_names),
826 )
827 )
828 else:
829 fix = ""
830
831 try:
832 field = through._model_meta.get_forward_field(field_name)
833 except FieldDoesNotExist:
834 errors.append(
835 PreflightResult(
836 fix=f"The intermediary model '{qualified_model_name}' has no field '{field_name}'. {fix}",
837 obj=self,
838 id="fields.m2m_through_field_not_found",
839 )
840 )
841 else:
842 if not (
843 isinstance(field, RelatedField)
844 and field.remote_field.model_ref == related_model
845 ):
846 errors.append(
847 PreflightResult(
848 fix=f"'{through.model_options.object_name}.{field_name}' is not a foreign key to '{related_model_name}'. {fix}",
849 obj=self,
850 id="fields.m2m_through_field_not_fk_to_model",
851 )
852 )
853
854 return errors
855
856 def _check_table_uniqueness(self, **kwargs: Any) -> list[PreflightResult]:
857 if isinstance(self.remote_field.through_ref, str):
858 return []
859 registered_tables = {
860 model.model_options.db_table: model
861 for model in self.meta.models_registry.get_models()
862 if model != self.remote_field.through
863 }
864 m2m_db_table = self.m2m_db_table()
865 model = registered_tables.get(m2m_db_table)
866 # Check if there's already a m2m field using the same through model.
867 if model and model != self.remote_field.through:
868 clashing_obj = model.model_options.label
869 return [
870 PreflightResult(
871 fix=(
872 f"The field's intermediary table '{m2m_db_table}' clashes with the "
873 f"table name of '{clashing_obj}'. "
874 "Change the through model's db_table or use a different model."
875 ),
876 obj=self,
877 id="fields.m2m_table_name_clash",
878 )
879 ]
880 return []
881
882 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
883 name, path, args, kwargs = super().deconstruct()
884
885 # Lowercase model names as they should be treated as case-insensitive.
886 if isinstance(self.remote_field.model_ref, str):
887 if "." in self.remote_field.model_ref:
888 package_label, model_name = self.remote_field.model_ref.split(".")
889 kwargs["to"] = f"{package_label}.{model_name.lower()}"
890 else:
891 kwargs["to"] = self.remote_field.model_ref.lower()
892 else:
893 kwargs["to"] = self.remote_field.model_ref.model_options.label_lower
894
895 if isinstance(self.remote_field.through_ref, str):
896 kwargs["through"] = self.remote_field.through_ref
897 else:
898 kwargs["through"] = self.remote_field.through_ref.model_options.label
899
900 return name, path, args, kwargs
901
902 def _get_path_info(self, direct: bool = False) -> list[PathInfo]:
903 """Called by both direct and indirect m2m traversal."""
904 int_model = self.remote_field.through
905 # M2M through model fields are always ForeignKey
906 linkfield1 = cast(
907 ForeignKeyField,
908 int_model._model_meta.get_forward_field(self.m2m_field_name()),
909 )
910 linkfield2 = cast(
911 ForeignKeyField,
912 int_model._model_meta.get_forward_field(self.m2m_reverse_field_name()),
913 )
914 if direct:
915 join1infos = linkfield1.reverse_path_infos
916 join2infos = linkfield2.path_infos
917 else:
918 join1infos = linkfield2.reverse_path_infos
919 join2infos = linkfield1.path_infos
920
921 return [*join1infos, *join2infos]
922
923 @cached_property
924 def path_infos(self) -> list[PathInfo]:
925 return self._get_path_info(direct=True)
926
927 @cached_property
928 def reverse_path_infos(self) -> list[PathInfo]:
929 return self._get_path_info(direct=False)
930
931 def _get_m2m_db_table(self) -> str:
932 """
933 Function that can be curried to provide the m2m table name for this
934 relation.
935 """
936 return self.remote_field.through.model_options.db_table
937
938 def _get_m2m_attr(self, related: Any, attr: str) -> Any:
939 """
940 Function that can be curried to provide the source accessor or DB
941 column name for the m2m table.
942 """
943 cache_attr = f"_m2m_{attr}_cache"
944 if hasattr(self, cache_attr):
945 return getattr(self, cache_attr)
946 if self.remote_field.through_fields is not None:
947 link_field_name: str | None = self.remote_field.through_fields[0]
948 else:
949 link_field_name = None
950 for f in self.remote_field.through._model_meta.fields:
951 if (
952 isinstance(f, RelatedField)
953 and f.remote_field.model == related.related_model
954 and (link_field_name is None or link_field_name == f.name)
955 ):
956 setattr(self, cache_attr, getattr(f, attr))
957 return getattr(self, cache_attr)
958 return None
959
960 def _get_m2m_reverse_attr(self, related: Any, attr: str) -> Any:
961 """
962 Function that can be curried to provide the related accessor or DB
963 column name for the m2m table.
964 """
965 cache_attr = f"_m2m_reverse_{attr}_cache"
966 if hasattr(self, cache_attr):
967 return getattr(self, cache_attr)
968 found = False
969 if self.remote_field.through_fields is not None:
970 link_field_name: str | None = self.remote_field.through_fields[1]
971 else:
972 link_field_name = None
973 for f in self.remote_field.through._model_meta.fields:
974 if isinstance(f, RelatedField) and f.remote_field.model == related.model:
975 if link_field_name is None and related.related_model == related.model:
976 # If this is an m2m-intermediate to self,
977 # the first foreign key you find will be
978 # the source column. Keep searching for
979 # the second foreign key.
980 if found:
981 setattr(self, cache_attr, getattr(f, attr))
982 break
983 else:
984 found = True
985 elif link_field_name is None or link_field_name == f.name:
986 setattr(self, cache_attr, getattr(f, attr))
987 break
988 return getattr(self, cache_attr)
989
990 def contribute_to_class(self, cls: type[Model], name: str) -> None:
991 super().contribute_to_class(cls, name)
992
993 def resolve_through_model(
994 _: Any, model: type[Model], field: ManyToManyField
995 ) -> None:
996 field.remote_field.through_ref = model
997
998 lazy_related_operation(
999 resolve_through_model,
1000 cls,
1001 self.remote_field.through_ref,
1002 field=self,
1003 )
1004
1005 # Add the descriptor for the m2m relation.
1006 setattr(cls, self.name, ForwardManyToManyDescriptor(self.remote_field))
1007
1008 # Set up the accessor for the m2m table name for the relation.
1009 self.m2m_db_table = self._get_m2m_db_table
1010
1011 def do_related_class(self, other: type[Model], cls: type[Model]) -> None:
1012 """Set up M2M metadata accessors for the through table."""
1013 super().do_related_class(other, cls)
1014
1015 # Set up the accessors for the column names on the m2m table.
1016 # These are used during query construction and schema operations.
1017 related = self.remote_field
1018 self.m2m_column_name = partial(self._get_m2m_attr, related, "column")
1019 self.m2m_reverse_name = partial(self._get_m2m_reverse_attr, related, "column")
1020
1021 self.m2m_field_name = partial(self._get_m2m_attr, related, "name")
1022 self.m2m_reverse_field_name = partial(
1023 self._get_m2m_reverse_attr, related, "name"
1024 )
1025
1026 def set_attributes_from_rel(self) -> None:
1027 pass
1028
1029 def value_from_object(self, obj: Model) -> list[Any]:
1030 return [] if obj.id is None else list(getattr(obj, self.name).query)
1031
1032 def save_form_data(self, instance: Model, data: Any) -> None:
1033 getattr(instance, self.name).set(data)
1034
1035 def db_type(self) -> None:
1036 # A ManyToManyField is not represented by a single column,
1037 # so return None.
1038 return None