Docs plain-postgres plain postgres fields reverse_related.py
Raw LLM
"""
"Rel objects" for related fields.

"Rel objects" (for lack of a better name) carry information about the relation
modeled by a related field and provide some utility functions. They're stored
in the ``remote_field`` attribute of the field.

They also act as reverse fields for the purposes of the Meta API because
they're the closest concept currently available.
"""

from functools import cached_property
from typing import TYPE_CHECKING, Any

from plain.utils.hashable import make_hashable

from .mixins import FieldCacheMixin

if TYPE_CHECKING:
    from plain.postgres.base import Model
    from plain.postgres.deletion import OnDelete
    from plain.postgres.fields import Field
    from plain.postgres.fields.related import (
        ForeignKeyField,
        ManyToManyField,
        RelatedField,
    )
    from plain.postgres.lookups import Lookup
    from plain.postgres.query_utils import PathInfo


class ForeignObjectRel(FieldCacheMixin):
    """
    Used by ForeignKeyField to store information about the relation.

    ``_model_meta.get_fields()`` returns this class to provide access to the field
    flags for the reverse relation.
    """

    # Field flags
    auto_created = True

    # Reverse relations are always nullable (Plain can't enforce that a
    # foreign key on the related model points to this model).
    allow_null = True
    empty_strings_allowed = False

    # Type annotations for instance attributes
    on_delete: OnDelete | None
    # The target model as given to the field: a "package.Model" string (or
    # "self") until lazy_related_operation resolves it to the class. Read
    # `model` for the resolved class.
    model_ref: str | type[Model]

    def __init__(
        self,
        *,
        field: RelatedField,
        to: str | type[Model],
        related_query_name: str | None = None,
        on_delete: OnDelete | None = None,
    ):
        # Annotated here rather than as a class-body attribute: RelatedField
        # is itself a descriptor, so a class-body `field: RelatedField` would
        # route every `self.field` read through its __get__. ForeignKeyRel
        # and ManyToManyRel below narrow this further, the same way.
        self.field: RelatedField = field
        self.model_ref = to
        self.related_query_name = related_query_name
        self.on_delete = on_delete

        self.symmetrical = False

    # Some of the following cached_properties can't be initialized in
    # __init__ as the field doesn't have its model yet. Calling these methods
    # before field.contribute_to_class() has been called will result in
    # AttributeError
    @property
    def model(self) -> type[Model]:
        """The resolved target model class."""
        # Deliberately not AttributeError: getattr(rel, "model", None) and
        # hasattr() would swallow that and silently report "no model".
        if isinstance(self.model_ref, str):
            raise TypeError(
                f"Related model {self.model_ref!r} for {self.field!r} has not been "
                "resolved yet; use model_ref to read the unresolved reference."
            )
        return self.model_ref

    @cached_property
    def name(self) -> str:
        return self.field.related_query_name()

    @property
    def remote_field(self) -> RelatedField:
        return self.field

    @property
    def target_field(self) -> Field:
        """
        The remote field this relation filters against. Read dynamically by the
        related-lookup prep path (`output_field.target_field`) when the relation
        itself is the lookup target, e.g. ``Parent.query.filter(child=obj)``.
        """
        return self.path_infos[-1].target_field

    @cached_property
    def related_model(self) -> type[Model]:
        if not self.field.model:
            raise AttributeError(
                "This property can't be accessed before self.field.contribute_to_class "
                "has been called."
            )
        return self.field.model

    def get_lookup(self, lookup: str) -> type[Lookup] | None:
        return self.field.get_lookup(lookup)

    def __repr__(self) -> str:
        return f"<{type(self).__name__}: {self.related_model.model_options.package_label}.{self.related_model.model_options.model_name}>"

    @property
    def identity(self) -> tuple[Any, ...]:
        return (
            self.field,
            self.model_ref,
            self.related_query_name,
            self.on_delete,
            self.symmetrical,
        )

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, self.__class__):
            return NotImplemented
        return self.identity == other.identity

    def __hash__(self) -> int:
        return hash(self.identity)

    def __getstate__(self) -> dict[str, Any]:
        state = self.__dict__.copy()
        # Delete the path_infos cached property because it can be recalculated
        # at first invocation after deserialization. The attribute must be
        # removed because subclasses like ForeignKeyRel may have a PathInfo
        # which contains an intermediate M2M table that's been dynamically
        # created and doesn't exist in the .models module.
        # This is a reverse relation, so there is no reverse_path_infos to
        # delete.
        state.pop("path_infos", None)
        return state

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

    def get_cache_name(self) -> str:
        """
        Return the name of the cache key to use for storing an instance of the
        forward model on the reverse model.

        Uses the related_query_name for caching, which provides a stable name
        for prefetch() operations.
        """
        return self.field.related_query_name()


class ForeignKeyRel(ForeignObjectRel):
    """
    Used by the ForeignKeyField field to store information about the relation.

    ``_model_meta.get_fields()`` returns this class to provide access to the
    reverse relation. Use ``isinstance(rel, ForeignKeyRel)`` to identify
    one-to-many reverse relations.
    """

    # Type annotations for instance attributes
    on_delete: OnDelete  # narrowed: FK rel always has a concrete action

    def __init__(
        self,
        *,
        field: ForeignKeyField,
        to: str | type[Model],
        on_delete: OnDelete,
        related_query_name: str | None = None,
    ):
        super().__init__(
            field=field,
            to=to,
            related_query_name=related_query_name,
            on_delete=on_delete,
        )
        # Narrows self.field from RelatedField to ForeignKeyField (see
        # ForeignObjectRel.__init__ for why this can't be a class-body
        # annotation).
        self.field: ForeignKeyField

    def __getstate__(self) -> dict[str, Any]:
        state = super().__getstate__()
        state.pop("related_model", None)
        return state

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


class ManyToManyRel(ForeignObjectRel):
    """
    Used by ManyToManyField to store information about the relation.

    ``_model_meta.get_fields()`` returns this class to provide access to the field
    flags for the reverse relation.
    """

    # Type annotations for instance attributes
    through_fields: tuple[str, str] | None
    # The intermediary model as given to the field; a string until
    # lazy_related_operation resolves it. Read `through` for the class.
    through_ref: str | type[Model]

    def __init__(
        self,
        *,
        field: ManyToManyField,
        to: str | type[Model],
        through: str | type[Model],
        through_fields: tuple[str, str] | None = None,
        related_query_name: str | None = None,
        symmetrical: bool = True,
    ):
        super().__init__(
            field=field,
            to=to,
            related_query_name=related_query_name,
        )
        # Narrows self.field from RelatedField to ManyToManyField (see
        # ForeignObjectRel.__init__ for why this can't be a class-body
        # annotation).
        self.field: ManyToManyField

        self.through_ref = through
        self.through_fields = through_fields

        self.symmetrical = symmetrical

    @property
    def identity(self) -> tuple[Any, ...]:
        return super().identity + (
            self.through_ref,
            make_hashable(self.through_fields),
        )

    @property
    def through(self) -> type[Model]:
        """The resolved intermediary model class."""
        # Deliberately not AttributeError, for the same reason as `model`.
        if isinstance(self.through_ref, str):
            raise TypeError(
                f"Through model {self.through_ref!r} for {self.field!r} has not been "
                "resolved yet; use through_ref to read the unresolved reference."
            )
        return self.through_ref
On this page