Docs plain-postgres plain postgres fields related.py
Raw LLM
from __future__ import annotations

import copy
import operator
from collections.abc import Callable, Sequence
from functools import cached_property, partial
from typing import TYPE_CHECKING, Any, Self, cast

from plain import exceptions
from plain.postgres.constants import LOOKUP_SEP
from plain.postgres.deletion import NO_ACTION, SET_NULL, OnDelete
from plain.postgres.exceptions import FieldDoesNotExist, FieldError
from plain.postgres.query_utils import PathInfo, Q
from plain.postgres.utils import make_model_tuple
from plain.preflight import PreflightResult

from ..registry import models_registry
from . import BLANK_CHOICE_DASH, Field
from .base import ColumnField
from .mixins import FieldCacheMixin
from .related_descriptors import (
    ForwardForeignKeyDescriptor,
    ForwardManyToManyDescriptor,
)
from .related_lookups import (
    RelatedExact,
    RelatedGreaterThan,
    RelatedGreaterThanOrEqual,
    RelatedIn,
    RelatedIsNull,
    RelatedLessThan,
    RelatedLessThanOrEqual,
)
from .reverse_related import ForeignKeyRel, ManyToManyRel

if TYPE_CHECKING:
    from plain.postgres.base import Model
    from plain.postgres.connection import DatabaseConnection
    from plain.postgres.fields.reverse_related import ForeignObjectRel

RECURSIVE_RELATIONSHIP_CONSTANT = "self"


def resolve_relation(
    scope_model: type[Model], relation: type[Model] | str
) -> type[Model] | str:
    """
    Transform relation into a model or fully-qualified model string of the form
    "package_label.ModelName", relative to scope_model.

    The relation argument can be:
      * RECURSIVE_RELATIONSHIP_CONSTANT, i.e. the string "self", in which case
        the model argument will be returned.
      * A bare model name without an package_label, in which case scope_model's
        package_label will be prepended.
      * An "package_label.ModelName" string.
      * A model class, which will be returned unchanged.
    """
    # Check for recursive relations
    if relation == RECURSIVE_RELATIONSHIP_CONSTANT:
        relation = scope_model

    # Look for an "app.Model" relation
    if isinstance(relation, str):
        if "." not in relation:
            relation = f"{scope_model.model_options.package_label}.{relation}"

    return relation


def lazy_related_operation(
    function: Any, model: type[Model], *related_models: type[Model] | str, **kwargs: Any
) -> None:
    """
    Schedule `function` to be called once `model` and all `related_models`
    have been imported and registered with the app registry. `function` will
    be called with the newly-loaded model classes as its positional arguments,
    plus any optional keyword arguments.

    The `model` argument must be a model class. Each subsequent positional
    argument is another model, or a reference to another model - see
    `resolve_relation()` for the various forms these may take. Any relative
    references will be resolved relative to `model`.

    This is a convenience wrapper for `Packages.lazy_model_operation` - the app
    registry model used is the one found in `model._model_meta.models_registry`.
    """
    models = [model] + [resolve_relation(model, rel) for rel in related_models]
    model_keys = (make_model_tuple(m) for m in models)
    models_registry = model._model_meta.models_registry
    return models_registry.lazy_model_operation(
        partial(function, **kwargs), *model_keys
    )


class RelatedField(FieldCacheMixin, Field):
    """Base class that all relational fields inherit from."""

    non_migration_attrs = (
        *Field.non_migration_attrs,
        "limit_choices_to",
        "related_query_name",
    )

    # RelatedField always has a remote_field (never None)
    remote_field: ForeignObjectRel
    # path_infos is implemented as @cached_property in subclasses (ForeignKey, ManyToManyField)
    path_infos: list[PathInfo]
    # Set by ForeignKeyField / ManyToManyField in their __init__; declared
    # here so RelatedField methods (deconstruct, related_query_name) can
    # reference them without isinstance-narrowing.
    _related_query_name: str | None
    _limit_choices_to: Any

    # No __init__: ForeignKeyField and ManyToManyField each set
    # _related_query_name, _limit_choices_to, and remote_field themselves.

    def __deepcopy__(self, memodict: dict[int, Any]) -> Self:
        # Handle remote_field deepcopy for RelatedFields
        obj = super().__deepcopy__(memodict)
        obj.remote_field = copy.copy(self.remote_field)
        if hasattr(self.remote_field, "field") and self.remote_field.field is self:
            obj.remote_field.field = obj  # ty: ignore[invalid-assignment]
        return obj

    @cached_property
    def related_model(self) -> type[Model]:
        # Can't cache this property until all the models are loaded.
        models_registry.check_ready()
        return self.remote_field.model

    def preflight(self, **kwargs: Any) -> list[PreflightResult]:
        return [
            *super().preflight(**kwargs),
            *self._check_related_query_name_is_valid(),
            *self._check_relation_model_exists(),
            *self._check_clashes(),
        ]

    def _check_related_query_name_is_valid(self) -> list[PreflightResult]:
        # Always validate related_query_name since it's still used for ORM queries
        # (e.g., User.query.filter(articles__title="..."))
        rel_query_name = self.related_query_name()
        errors: list[PreflightResult] = []
        if rel_query_name.endswith("_"):
            errors.append(
                PreflightResult(
                    fix=(
                        f"Reverse query name '{rel_query_name}' must not end with an underscore. "
                        "Use a different related_query_name."
                    ),
                    obj=self,
                    id="fields.related_field_accessor_clash",
                )
            )
        if LOOKUP_SEP in rel_query_name:
            errors.append(
                PreflightResult(
                    fix=(
                        f"Reverse query name '{rel_query_name}' must not contain '{LOOKUP_SEP}'. "
                        "Use a different related_query_name."
                    ),
                    obj=self,
                    id="fields.related_field_query_name_clash",
                )
            )
        return errors

    def _check_relation_model_exists(self) -> list[PreflightResult]:
        rel_is_missing = (
            self.remote_field.model not in self.meta.models_registry.get_models()
        )
        rel_is_string = isinstance(self.remote_field.model, str)
        model_name = (
            self.remote_field.model
            if rel_is_string
            else self.remote_field.model.model_options.object_name
        )
        if rel_is_missing and rel_is_string:
            return [
                PreflightResult(
                    fix=(
                        f"Field defines a relation with model '{model_name}', which is not "
                        "installed. Ensure the model's package is registered."
                    ),
                    obj=self,
                    id="fields.related_model_not_installed",
                )
            ]
        return []

    def _check_clashes(self) -> list[PreflightResult]:
        """Check accessor and reverse query name clashes."""
        from plain.postgres.base import ModelBase

        errors: list[PreflightResult] = []

        # f.remote_field.model may be a string instead of a model. Skip if
        # model name is not resolved.
        if not isinstance(self.remote_field.model, ModelBase):
            return []

        # Consider that we are checking field `Model.foreign` and the models
        # are:
        #
        #     class Target(models.Model):
        #         model = models.IntegerField()
        #         model_set = models.IntegerField()
        #
        #     class Model(models.Model):
        #         foreign = models.ForeignKeyField(Target)
        #         m2m = models.ManyToManyField(Target)

        # rel_options.object_name == "Target"
        rel_meta = self.remote_field.model._model_meta
        rel_options = self.remote_field.model.model_options
        rel_query_name = self.related_query_name()  # i. e. "model"
        # i.e. "package_label.Model.field".
        field_name = f"{self.model.model_options.label}.{self.name}"

        # Check clashes between reverse query name of `field`
        # and any other field name.
        potential_clashes = rel_meta.fields + rel_meta.many_to_many
        for clash_field in potential_clashes:
            # i.e. "package_label.Target.model_set".
            clash_name = f"{rel_options.label}.{clash_field.name}"
            if clash_field.name == rel_query_name:
                errors.append(
                    PreflightResult(
                        fix=(
                            f"Reverse query name for '{field_name}' clashes with field name '{clash_name}'. "
                            f"Rename field '{clash_name}' or use a different related_query_name."
                        ),
                        obj=self,
                        id="fields.related_accessor_clash_manager",
                    )
                )

        return errors

    def db_type(self) -> str | None:
        # By default related field will not have a column as it relates to
        # columns from another table.
        return None

    def unqualified_db_type(self) -> str | None:
        return self.rel_db_type()

    def contribute_to_class(self, cls: type[Model], name: str) -> None:
        super().contribute_to_class(cls, name)

        self.meta = cls._model_meta

        if self.remote_field.related_query_name:
            related_query_name = self.remote_field.related_query_name % {
                "class": cls.__name__.lower(),
                "package_label": cls.model_options.package_label.lower(),
            }
            self.remote_field.related_query_name = related_query_name

        def resolve_related_class(
            model: type[Model], related: type[Model], field: RelatedField
        ) -> None:
            field.remote_field.model = related
            field.do_related_class(related, model)

        lazy_related_operation(
            resolve_related_class,
            cls,
            self.remote_field.model,
            field=self,
        )

    def deconstruct(self) -> tuple[str | None, str, list[Any], dict[str, Any]]:
        name, path, args, kwargs = super().deconstruct()
        if self._limit_choices_to:
            kwargs["limit_choices_to"] = self._limit_choices_to
        if self._related_query_name is not None:
            kwargs["related_query_name"] = self._related_query_name
        return name, path, args, kwargs

    def set_attributes_from_rel(self) -> None:
        self.name = self.name or (
            self.remote_field.model.model_options.model_name + "_" + "id"
        )
        self.remote_field.set_field_name()

    def do_related_class(self, other: type[Model], cls: type[Model]) -> None:
        self.set_attributes_from_rel()

    def get_limit_choices_to(self) -> Any:
        """
        Return ``limit_choices_to`` for this model field.

        If it is a callable, it will be invoked and the result will be
        returned.
        """
        if callable(self.remote_field.limit_choices_to):
            return self.remote_field.limit_choices_to()  # ty: ignore[call-top-callable]
        return self.remote_field.limit_choices_to

    def get_choices(
        self,
        include_blank: bool = True,
        blank_choice: list[tuple[str, str]] = BLANK_CHOICE_DASH,
        limit_choices_to: Any = None,
        ordering: tuple[str, ...] = (),
    ) -> list[tuple[Any, str]]:
        """Return choices from the related model, for use as <select> options."""
        rel_model = self.remote_field.model
        if rel_model is None:
            return blank_choice if include_blank else []
        limit_choices_to = limit_choices_to or self.get_limit_choices_to()
        get_related_field = getattr(self.remote_field, "get_related_field", None)
        related_field_name = (
            get_related_field().attname if get_related_field is not None else "id"
        )
        choice_func = operator.attrgetter(related_field_name)
        qs = rel_model.query.complex_filter(limit_choices_to)
        if ordering:
            qs = qs.order_by(*ordering)
        return (blank_choice if include_blank else []) + [
            (choice_func(x), str(x)) for x in qs
        ]

    def related_query_name(self) -> str:
        """
        Define the name that can be used to identify this related object in a
        table-spanning query.
        """
        return (
            self.remote_field.related_query_name or self.model.model_options.model_name
        )

    @property
    def target_field(self) -> Field:
        """
        When filtering against this relation, return the field on the remote
        model against which the filtering should happen.
        """
        target_fields = self.path_infos[-1].target_fields
        if len(target_fields) > 1:
            raise FieldError(
                "The relation has multiple target fields, but only single target field "
                "was asked for"
            )
        return target_fields[0]

    def get_cache_name(self) -> str:
        assert self.name is not None, "Field name must be set"
        return self.name


class ForeignKeyField(ColumnField, RelatedField):
    # Narrow the base class's `ForeignObjectRel` annotation — a FK's remote_field
    # is always a ForeignKeyRel with a concrete on_delete action.
    remote_field: ForeignKeyRel

    """
    Provide a many-to-one relation by adding a column to the local model
    to hold the remote value.

    ForeignKeyField targets the primary key (id) of the remote model.
    """

    non_migration_attrs = (
        *RelatedField.non_migration_attrs,
        *ColumnField.non_migration_attrs,
        "on_delete",
    )

    empty_strings_allowed = False

    def __init__(
        self,
        to: type[Model] | str,
        on_delete: OnDelete,
        related_query_name: str | None = None,
        limit_choices_to: Any = None,
        db_constraint: bool = True,
        *,
        required: bool = True,
        allow_null: bool = False,
        validators: Sequence[Callable[..., Any]] = (),
    ):
        # `default` and `choices` are intentionally not accepted: a hardcoded
        # FK id default is a portability/existence footgun, and the related
        # model itself already defines the valid set. Use `limit_choices_to`
        # to constrain the target rows.
        if not isinstance(to, str):
            try:
                to.model_options.model_name
            except AttributeError:
                raise TypeError(
                    f"{self.__class__.__name__}({to!r}) is invalid. First parameter to ForeignKeyField must be "
                    f"either a model, a model name, or the string {RECURSIVE_RELATIONSHIP_CONSTANT!r}"
                )
        if not isinstance(on_delete, OnDelete):
            raise TypeError(
                "on_delete must be one of plain.postgres.CASCADE, SET_NULL, "
                f"RESTRICT, or NO_ACTION; got {on_delete!r}"
            )

        super().__init__(
            required=required,
            allow_null=allow_null,
            validators=validators,
        )
        self._related_query_name = related_query_name
        self._limit_choices_to = limit_choices_to
        self.remote_field = ForeignKeyRel(
            field=self,
            to=to,
            on_delete=on_delete,
            related_query_name=related_query_name,
            limit_choices_to=limit_choices_to,
        )
        self.db_constraint = db_constraint

    def __copy__(self) -> ForeignKeyField:
        obj = super().__copy__()
        # Remove any cached PathInfo values.
        obj.__dict__.pop("path_infos", None)
        obj.__dict__.pop("reverse_path_infos", None)
        return obj

    def __set__(self, instance: Any, value: Any) -> None:
        """
        Override Field's __set__ to clear cached related object when FK value changes.

        This ensures that when you change obj.user_id, the cached obj.user is invalidated.
        """
        # Check if value is changing and clear cache if needed
        if (
            hasattr(self, "attname")
            and instance.__dict__.get(self.attname) != value
            and self.is_cached(instance)
        ):
            self.delete_cached_value(instance)

        # Call parent's __set__ to do the actual assignment
        super().__set__(instance, value)

    @cached_property
    def related_fields(self) -> list[tuple[ForeignKeyField, Field]]:
        return self.resolve_related_fields()

    @cached_property
    def reverse_related_fields(self) -> list[tuple[Field, Field]]:
        return [(rhs_field, lhs_field) for lhs_field, rhs_field in self.related_fields]

    @cached_property
    def local_related_fields(self) -> tuple[Field, ...]:
        return tuple(lhs_field for lhs_field, rhs_field in self.related_fields)

    @cached_property
    def foreign_related_fields(self) -> tuple[Field, ...]:
        return tuple(
            rhs_field for lhs_field, rhs_field in self.related_fields if rhs_field
        )

    def get_forward_related_filter(self, obj: Model) -> dict[str, Any]:
        """
        Return the keyword arguments that when supplied to
        self.model.object.filter(), would select all instances related through
        this field to the remote obj. This is used to build the querysets
        returned by related descriptors. obj is an instance of
        self.related_field.model.
        """
        return {
            f"{self.name}__{rh_field.name}": getattr(obj, rh_field.attname)
            for _, rh_field in self.related_fields
        }

    def get_reverse_related_filter(self, obj: Model) -> Q:
        """
        Complement to get_forward_related_filter(). Return the keyword
        arguments that when passed to self.related_field.model.object.filter()
        select all instances of self.related_field.model related through
        this field to obj. obj is an instance of self.model.
        """
        return Q.create(
            [
                (rh_field.attname, getattr(obj, lh_field.attname))
                for lh_field, rh_field in self.related_fields
            ]
        )

    def get_local_related_value(self, instance: Model) -> tuple[Any, ...]:
        # Always returns the value of the single local field
        field = self.local_related_fields[0]
        if field.primary_key:
            return (instance.id,)
        return (getattr(instance, field.attname),)

    def get_foreign_related_value(self, instance: Model) -> tuple[Any, ...]:
        # Always returns the id of the foreign instance
        return (instance.id,)

    def get_joining_columns(
        self, reverse_join: bool = False
    ) -> tuple[tuple[str, str], ...]:
        # Always returns a single column pair
        if reverse_join:
            from_field, to_field = self.related_fields[0]
            return ((to_field.column, from_field.column),)
        else:
            from_field, to_field = self.related_fields[0]
            return ((from_field.column, to_field.column),)

    def get_reverse_joining_columns(self) -> tuple[tuple[str, str], ...]:
        return self.get_joining_columns(reverse_join=True)

    def get_path_info(self, filtered_relation: Any = None) -> list[PathInfo]:
        """Get path from this field to the related model."""
        meta = self.remote_field.model._model_meta
        from_meta = self.model._model_meta
        return [
            PathInfo(
                from_meta=from_meta,
                to_meta=meta,
                target_fields=self.foreign_related_fields,
                join_field=self,
                m2m=False,
                direct=True,
                filtered_relation=filtered_relation,
            )
        ]

    @cached_property
    def path_infos(self) -> list[PathInfo]:
        return self.get_path_info()

    def get_reverse_path_info(self, filtered_relation: Any = None) -> list[PathInfo]:
        """Get path from the related model to this field's model."""
        meta = self.model._model_meta
        from_meta = self.remote_field.model._model_meta
        return [
            PathInfo(
                from_meta=from_meta,
                to_meta=meta,
                target_fields=(meta.get_forward_field("id"),),
                join_field=self.remote_field,
                m2m=not self.primary_key,
                direct=False,
                filtered_relation=filtered_relation,
            )
        ]

    @cached_property
    def reverse_path_infos(self) -> list[PathInfo]:
        return self.get_reverse_path_info()

    def contribute_to_class(self, cls: type[Model], name: str) -> None:
        super().contribute_to_class(cls, name)
        setattr(cls, name, ForwardForeignKeyDescriptor(self))

    def preflight(self, **kwargs: Any) -> list[PreflightResult]:
        return [
            *super().preflight(**kwargs),
            *self._check_on_delete(),
        ]

    def _check_on_delete(self) -> list[PreflightResult]:
        on_delete = getattr(self.remote_field, "on_delete", None)
        results: list[PreflightResult] = []
        if on_delete is SET_NULL and not self.allow_null:
            results.append(
                PreflightResult(
                    fix=(
                        "Field specifies on_delete=SET_NULL, but cannot be null. "
                        "Set allow_null=True argument on the field, or change the on_delete rule."
                    ),
                    obj=self,
                    id="fields.foreign_key_null_constraint_violation",
                )
            )
        if not self.db_constraint and on_delete is not NO_ACTION:
            results.append(
                PreflightResult(
                    fix=(
                        "db_constraint=False requires on_delete=NO_ACTION. "
                        "Without a FOREIGN KEY constraint there is no place for "
                        "Postgres to apply the delete behavior."
                    ),
                    obj=self,
                    id="fields.foreign_key_unconstrained_requires_no_action",
                )
            )
        return results

    def deconstruct(self) -> tuple[str | None, str, list[Any], dict[str, Any]]:
        name, path, args, kwargs = super().deconstruct()
        kwargs["on_delete"] = self.remote_field.on_delete

        if isinstance(self.remote_field.model, str):
            if "." in self.remote_field.model:
                package_label, model_name = self.remote_field.model.split(".")
                kwargs["to"] = f"{package_label}.{model_name.lower()}"
            else:
                kwargs["to"] = self.remote_field.model.lower()
        else:
            kwargs["to"] = self.remote_field.model.model_options.label_lower

        if self.db_constraint is not True:
            kwargs["db_constraint"] = self.db_constraint

        return name, path, args, kwargs

    def to_python(self, value: Any) -> Any:
        return self.target_field.to_python(value)

    @property
    def target_field(self) -> Field:
        return self.foreign_related_fields[0]

    def validate(self, value: Any, model_instance: Model) -> None:
        super().validate(value, model_instance)
        if value is None:
            return None

        field_name = self.remote_field.field_name
        if field_name is None:
            raise ValueError("remote_field.field_name cannot be None")
        qs = self.remote_field.model._model_meta.base_queryset.filter(
            **{field_name: value}
        )
        qs = qs.complex_filter(self.get_limit_choices_to())
        if not qs.exists():
            raise exceptions.ValidationError(
                "%(model)s instance with %(field)s %(value)r does not exist.",
                code="invalid",
                params={
                    "model": self.remote_field.model.model_options.model_name,
                    "id": value,
                    "field": self.remote_field.field_name,
                    "value": value,
                },
            )

    def resolve_related_fields(self) -> list[tuple[ForeignKeyField, Field]]:
        if isinstance(self.remote_field.model, str):
            raise ValueError(
                f"Related model {self.remote_field.model!r} cannot be resolved"
            )
        from_field = self
        to_field = self.remote_field.model._model_meta.get_forward_field("id")
        related_fields: list[tuple[ForeignKeyField, Field]] = [(from_field, to_field)]

        for from_field, to_field in related_fields:
            if to_field and to_field.model != self.remote_field.model:
                raise FieldError(
                    f"'{self.model.model_options.label}.{self.name}' refers to field '{to_field.name}' which is not local to model "
                    f"'{self.remote_field.model.model_options.label}'."
                )
        return related_fields

    def get_attname(self) -> str:
        return f"{self.name}_id"

    def get_db_prep_save(self, value: Any, connection: DatabaseConnection) -> Any:
        if value is None or (
            value == "" and not self.target_field.empty_strings_allowed
        ):
            return None
        else:
            return self.target_field.get_db_prep_save(value, connection=connection)

    def get_db_prep_value(
        self, value: Any, connection: DatabaseConnection, prepared: bool = False
    ) -> Any:
        return self.target_field.get_db_prep_value(value, connection, prepared)

    def get_prep_value(self, value: Any) -> Any:
        return self.target_field.get_prep_value(value)

    def db_type(self) -> str | None:
        return self.target_field.rel_db_type()

    def cast_db_type(self) -> str | None:
        return self.target_field.cast_db_type()

    def get_col(self, alias: str | None, output_field: Field | None = None) -> Any:
        if output_field is None:
            output_field = self.target_field
            while isinstance(output_field, ForeignKeyField):
                output_field = output_field.target_field
                if output_field is self:
                    raise ValueError("Cannot resolve output_field.")
        return super().get_col(alias, output_field)


# Register lookups for ForeignKey
ForeignKeyField.register_lookup(RelatedIn)
ForeignKeyField.register_lookup(RelatedExact)
ForeignKeyField.register_lookup(RelatedLessThan)
ForeignKeyField.register_lookup(RelatedGreaterThan)
ForeignKeyField.register_lookup(RelatedGreaterThanOrEqual)
ForeignKeyField.register_lookup(RelatedLessThanOrEqual)
ForeignKeyField.register_lookup(RelatedIsNull)


class ManyToManyField(RelatedField):
    """
    Provide a many-to-many relation by using an intermediary model that
    holds two ForeignKeyField fields pointed at the two sides of the relation.

    Unless a ``through`` model was provided, ManyToManyField will use the
    create_many_to_many_intermediary_model factory to automatically generate
    the intermediary model.
    """

    # ManyToManyField uses ManyToManyRel which has through/through_fields
    remote_field: ManyToManyRel

    def __init__(
        self,
        to: type[Model] | str,
        *,
        through: type[Model] | str,
        through_fields: tuple[str, str] | None = None,
        related_query_name: str | None = None,
        limit_choices_to: Any = None,
        symmetrical: bool | None = None,
    ):
        # M2M has no database column, so `required`, `allow_null`, `default`,
        # `validators`, and `choices` are intentionally not accepted. Membership
        # is managed through the related manager; filter target rows with
        # `limit_choices_to`.
        if not isinstance(to, str):
            try:
                to._model_meta
            except AttributeError:
                raise TypeError(
                    f"{self.__class__.__name__}({to!r}) is invalid. First parameter to ManyToManyField "
                    f"must be either a model, a model name, or the string {RECURSIVE_RELATIONSHIP_CONSTANT!r}"
                )

        if symmetrical is None:
            symmetrical = to == RECURSIVE_RELATIONSHIP_CONSTANT

        if not through:
            raise ValueError("ManyToManyField must have a 'through' argument.")

        self.remote_field = ManyToManyRel(
            field=self,
            to=to,
            related_query_name=related_query_name,
            limit_choices_to=limit_choices_to,
            symmetrical=symmetrical,
            through=through,
            through_fields=through_fields,
        )

        super().__init__()
        self._related_query_name = related_query_name
        self._limit_choices_to = limit_choices_to

    def preflight(self, **kwargs: Any) -> list[PreflightResult]:
        return [
            *super().preflight(**kwargs),
            *self._check_relationship_model(**kwargs),
            *self._check_table_uniqueness(**kwargs),
        ]

    def _check_relationship_model(
        self, from_model: type[Model] | None = None, **kwargs: Any
    ) -> list[PreflightResult]:
        if hasattr(self.remote_field.through, "_model_meta"):
            qualified_model_name = f"{self.remote_field.through.model_options.package_label}.{self.remote_field.through.__name__}"
        else:
            qualified_model_name = self.remote_field.through

        errors = []

        if self.remote_field.through not in self.meta.models_registry.get_models():
            # The relationship model is not installed.
            errors.append(
                PreflightResult(
                    fix=(
                        "Field specifies a many-to-many relation through model "
                        f"'{qualified_model_name}', which has not been installed. "
                        "Ensure the through model is properly defined and installed."
                    ),
                    obj=self,
                    id="fields.m2m_through_model_not_installed",
                )
            )

        else:
            assert from_model is not None, (
                "ManyToManyField with intermediate "
                "tables cannot be checked if you don't pass the model "
                "where the field is attached to."
            )
            # Set some useful local variables
            to_model = resolve_relation(from_model, self.remote_field.model)
            from_model_name = from_model.model_options.object_name
            if isinstance(to_model, str):
                to_model_name = to_model
            else:
                to_model_name = to_model.model_options.object_name
            relationship_model_name = (
                self.remote_field.through.model_options.object_name
            )
            self_referential = from_model == to_model
            # Count foreign keys in intermediate model
            if self_referential:
                seen_self = sum(
                    from_model == field.remote_field.model
                    for field in self.remote_field.through._model_meta.fields
                    if isinstance(field, RelatedField)
                )

                if seen_self > 2 and not self.remote_field.through_fields:
                    errors.append(
                        PreflightResult(
                            fix=(
                                "The model is used as an intermediate model by "
                                f"'{self}', but it has more than two foreign keys "
                                f"to '{from_model_name}', which is ambiguous. "
                                "Use through_fields to specify which two foreign keys "
                                "Plain should use."
                            ),
                            obj=self.remote_field.through,
                            id="fields.m2m_through_model_ambiguous_fks",
                        )
                    )

            else:
                # Count foreign keys in relationship model
                seen_from = sum(
                    from_model == field.remote_field.model
                    for field in self.remote_field.through._model_meta.fields
                    if isinstance(field, RelatedField)
                )
                seen_to = sum(
                    to_model == field.remote_field.model
                    for field in self.remote_field.through._model_meta.fields
                    if isinstance(field, RelatedField)
                )

                if seen_from > 1 and not self.remote_field.through_fields:
                    errors.append(
                        PreflightResult(
                            fix=(
                                "The model is used as an intermediate model by "
                                f"'{self}', but it has more than one foreign key "
                                f"from '{from_model_name}', which is ambiguous. You must specify "
                                "which foreign key Plain should use via the "
                                "through_fields keyword argument. "
                                "If you want to create a recursive relationship, "
                                f'use ManyToManyField("{RECURSIVE_RELATIONSHIP_CONSTANT}", through="{relationship_model_name}").'
                            ),
                            obj=self,
                            id="fields.m2m_through_model_invalid_recursive_from",
                        )
                    )

                if seen_to > 1 and not self.remote_field.through_fields:
                    errors.append(
                        PreflightResult(
                            fix=(
                                "The model is used as an intermediate model by "
                                f"'{self}', but it has more than one foreign key "
                                f"to '{to_model_name}', which is ambiguous. You must specify "
                                "which foreign key Plain should use via the "
                                "through_fields keyword argument. "
                                "If you want to create a recursive relationship, "
                                f'use ManyToManyField("{RECURSIVE_RELATIONSHIP_CONSTANT}", through="{relationship_model_name}").'
                            ),
                            obj=self,
                            id="fields.m2m_through_model_invalid_recursive_to",
                        )
                    )

                if seen_from == 0 or seen_to == 0:
                    errors.append(
                        PreflightResult(
                            fix=(
                                "The model is used as an intermediate model by "
                                f"'{self}', but it does not have a foreign key to '{from_model_name}' or '{to_model_name}'. "
                                "Add the required foreign keys to the through model."
                            ),
                            obj=self.remote_field.through,
                            id="fields.m2m_through_model_missing_fk",
                        )
                    )

        # Validate `through_fields`.
        if self.remote_field.through_fields is not None:
            # Validate that we're given an iterable of at least two items
            # and that none of them is "falsy".
            if not (
                len(self.remote_field.through_fields) >= 2
                and self.remote_field.through_fields[0]
                and self.remote_field.through_fields[1]
            ):
                errors.append(
                    PreflightResult(
                        fix=(
                            "Field specifies 'through_fields' but does not provide "
                            "the names of the two link fields that should be used "
                            f"for the relation through model '{qualified_model_name}'. "
                            "Make sure you specify 'through_fields' as "
                            "through_fields=('field1', 'field2')."
                        ),
                        obj=self,
                        id="fields.m2m_through_fields_wrong_length",
                    )
                )

            # Validate the given through fields -- they should be actual
            # fields on the through model, and also be foreign keys to the
            # expected models.
            else:
                assert from_model is not None, (
                    "ManyToManyField with intermediate "
                    "tables cannot be checked if you don't pass the model "
                    "where the field is attached to."
                )

                source, through, target = (
                    from_model,
                    self.remote_field.through,
                    self.remote_field.model,
                )
                source_field_name, target_field_name = self.remote_field.through_fields[
                    :2
                ]

                for field_name, related_model in (
                    (source_field_name, source),
                    (target_field_name, target),
                ):
                    possible_field_names = []
                    for f in through._model_meta.fields:
                        if (
                            hasattr(f, "remote_field")
                            and getattr(f.remote_field, "model", None) == related_model
                        ):
                            possible_field_names.append(f.name)
                    if possible_field_names:
                        fix = (
                            "Did you mean one of the following foreign keys to '{}': "
                            "{}?".format(
                                related_model.model_options.object_name,
                                ", ".join(possible_field_names),
                            )
                        )
                    else:
                        fix = ""

                    try:
                        field = through._model_meta.get_forward_field(field_name)
                    except FieldDoesNotExist:
                        errors.append(
                            PreflightResult(
                                fix=f"The intermediary model '{qualified_model_name}' has no field '{field_name}'. {fix}",
                                obj=self,
                                id="fields.m2m_through_field_not_found",
                            )
                        )
                    else:
                        if not (
                            isinstance(field, RelatedField)
                            and field.remote_field.model == related_model
                        ):
                            errors.append(
                                PreflightResult(
                                    fix=f"'{through.model_options.object_name}.{field_name}' is not a foreign key to '{related_model.model_options.object_name}'. {fix}",
                                    obj=self,
                                    id="fields.m2m_through_field_not_fk_to_model",
                                )
                            )

        return errors

    def _check_table_uniqueness(self, **kwargs: Any) -> list[PreflightResult]:
        if isinstance(self.remote_field.through, str):
            return []
        registered_tables = {
            model.model_options.db_table: model
            for model in self.meta.models_registry.get_models()
            if model != self.remote_field.through
        }
        m2m_db_table = self.m2m_db_table()
        model = registered_tables.get(m2m_db_table)
        # Check if there's already a m2m field using the same through model.
        if model and model != self.remote_field.through:
            clashing_obj = model.model_options.label
            return [
                PreflightResult(
                    fix=(
                        f"The field's intermediary table '{m2m_db_table}' clashes with the "
                        f"table name of '{clashing_obj}'. "
                        "Change the through model's db_table or use a different model."
                    ),
                    obj=self,
                    id="fields.m2m_table_name_clash",
                )
            ]
        return []

    def deconstruct(self) -> tuple[str | None, str, list[Any], dict[str, Any]]:
        name, path, args, kwargs = super().deconstruct()

        if self.remote_field.db_constraint is not True:
            kwargs["db_constraint"] = self.remote_field.db_constraint

        # Lowercase model names as they should be treated as case-insensitive.
        if isinstance(self.remote_field.model, str):
            if "." in self.remote_field.model:
                package_label, model_name = self.remote_field.model.split(".")
                kwargs["to"] = f"{package_label}.{model_name.lower()}"
            else:
                kwargs["to"] = self.remote_field.model.lower()
        else:
            kwargs["to"] = self.remote_field.model.model_options.label_lower

        if isinstance(self.remote_field.through, str):
            kwargs["through"] = self.remote_field.through
        else:
            kwargs["through"] = self.remote_field.through.model_options.label

        return name, path, args, kwargs

    def _get_path_info(
        self, direct: bool = False, filtered_relation: Any = None
    ) -> list[PathInfo]:
        """Called by both direct and indirect m2m traversal."""
        int_model = self.remote_field.through
        # M2M through model fields are always ForeignKey
        linkfield1 = cast(
            ForeignKeyField,
            int_model._model_meta.get_forward_field(self.m2m_field_name()),
        )
        linkfield2 = cast(
            ForeignKeyField,
            int_model._model_meta.get_forward_field(self.m2m_reverse_field_name()),
        )
        if direct:
            join1infos = linkfield1.reverse_path_infos
            if filtered_relation:
                join2infos = linkfield2.get_path_info(filtered_relation)
            else:
                join2infos = linkfield2.path_infos
        else:
            join1infos = linkfield2.reverse_path_infos
            if filtered_relation:
                join2infos = linkfield1.get_path_info(filtered_relation)
            else:
                join2infos = linkfield1.path_infos

        return [*join1infos, *join2infos]

    def get_path_info(self, filtered_relation: Any = None) -> list[PathInfo]:
        return self._get_path_info(direct=True, filtered_relation=filtered_relation)

    @cached_property
    def path_infos(self) -> list[PathInfo]:
        return self.get_path_info()

    def get_reverse_path_info(self, filtered_relation: Any = None) -> list[PathInfo]:
        return self._get_path_info(direct=False, filtered_relation=filtered_relation)

    @cached_property
    def reverse_path_infos(self) -> list[PathInfo]:
        return self.get_reverse_path_info()

    def _get_m2m_db_table(self) -> str:
        """
        Function that can be curried to provide the m2m table name for this
        relation.
        """
        return self.remote_field.through.model_options.db_table

    def _get_m2m_attr(self, related: Any, attr: str) -> Any:
        """
        Function that can be curried to provide the source accessor or DB
        column name for the m2m table.
        """
        cache_attr = f"_m2m_{attr}_cache"
        if hasattr(self, cache_attr):
            return getattr(self, cache_attr)
        if self.remote_field.through_fields is not None:
            link_field_name: str | None = self.remote_field.through_fields[0]
        else:
            link_field_name = None
        for f in self.remote_field.through._model_meta.fields:
            if (
                isinstance(f, RelatedField)
                and f.remote_field.model == related.related_model
                and (link_field_name is None or link_field_name == f.name)
            ):
                setattr(self, cache_attr, getattr(f, attr))
                return getattr(self, cache_attr)
        return None

    def _get_m2m_reverse_attr(self, related: Any, attr: str) -> Any:
        """
        Function that can be curried to provide the related accessor or DB
        column name for the m2m table.
        """
        cache_attr = f"_m2m_reverse_{attr}_cache"
        if hasattr(self, cache_attr):
            return getattr(self, cache_attr)
        found = False
        if self.remote_field.through_fields is not None:
            link_field_name: str | None = self.remote_field.through_fields[1]
        else:
            link_field_name = None
        for f in self.remote_field.through._model_meta.fields:
            if isinstance(f, RelatedField) and f.remote_field.model == related.model:
                if link_field_name is None and related.related_model == related.model:
                    # If this is an m2m-intermediate to self,
                    # the first foreign key you find will be
                    # the source column. Keep searching for
                    # the second foreign key.
                    if found:
                        setattr(self, cache_attr, getattr(f, attr))
                        break
                    else:
                        found = True
                elif link_field_name is None or link_field_name == f.name:
                    setattr(self, cache_attr, getattr(f, attr))
                    break
        return getattr(self, cache_attr)

    def contribute_to_class(self, cls: type[Model], name: str) -> None:
        super().contribute_to_class(cls, name)

        def resolve_through_model(
            _: Any, model: type[Model], field: ManyToManyField
        ) -> None:
            field.remote_field.through = model

        lazy_related_operation(
            resolve_through_model,
            cls,
            self.remote_field.through,
            field=self,
        )

        # Add the descriptor for the m2m relation.
        setattr(cls, self.name, ForwardManyToManyDescriptor(self.remote_field))  # ty: ignore[invalid-argument-type]

        # Set up the accessor for the m2m table name for the relation.
        self.m2m_db_table = self._get_m2m_db_table

    def do_related_class(self, other: type[Model], cls: type[Model]) -> None:
        """Set up M2M metadata accessors for the through table."""
        super().do_related_class(other, cls)

        # Set up the accessors for the column names on the m2m table.
        # These are used during query construction and schema operations.
        related = self.remote_field
        self.m2m_column_name = partial(self._get_m2m_attr, related, "column")
        self.m2m_reverse_name = partial(self._get_m2m_reverse_attr, related, "column")

        self.m2m_field_name = partial(self._get_m2m_attr, related, "name")
        self.m2m_reverse_field_name = partial(
            self._get_m2m_reverse_attr, related, "name"
        )

        get_m2m_rel = partial(self._get_m2m_attr, related, "remote_field")
        self.m2m_target_field_name = lambda: get_m2m_rel().field_name
        get_m2m_reverse_rel = partial(
            self._get_m2m_reverse_attr, related, "remote_field"
        )
        self.m2m_reverse_target_field_name = lambda: get_m2m_reverse_rel().field_name

    def set_attributes_from_rel(self) -> None:
        pass

    def value_from_object(self, obj: Model) -> list[Any]:
        return [] if obj.id is None else list(getattr(obj, self.attname).query)

    def save_form_data(self, instance: Model, data: Any) -> None:
        getattr(instance, self.attname).set(data)

    def db_type(self) -> None:
        # A ManyToManyField is not represented by a single column,
        # so return None.
        return None
On this page