1"""
  2"Rel objects" for related fields.
  3
  4"Rel objects" (for lack of a better name) carry information about the relation
  5modeled by a related field and provide some utility functions. They're stored
  6in the ``remote_field`` attribute of the field.
  7
  8They also act as reverse fields for the purposes of the Meta API because
  9they're the closest concept currently available.
 10"""
 11
 12from __future__ import annotations
 13
 14from functools import cached_property
 15from typing import TYPE_CHECKING, Any
 16
 17from plain.models.exceptions import FieldError
 18from plain.utils.hashable import make_hashable
 19
 20from . import BLANK_CHOICE_DASH
 21from .mixins import FieldCacheMixin
 22
 23if TYPE_CHECKING:
 24    from collections.abc import Callable
 25
 26    from plain.models.base import Model
 27    from plain.models.deletion import Collector
 28    from plain.models.fields import Field
 29    from plain.models.fields.related import (
 30        ForeignKeyField,
 31        ManyToManyField,
 32        RelatedField,
 33    )
 34    from plain.models.lookups import Lookup
 35    from plain.models.query_utils import PathInfo, Q
 36
 37    # Type alias for on_delete callbacks
 38    OnDeleteCallback = Callable[[Collector, Any, Any], None]
 39
 40
 41class ForeignObjectRel(FieldCacheMixin):
 42    """
 43    Used by ForeignKeyField to store information about the relation.
 44
 45    ``_model_meta.get_fields()`` returns this class to provide access to the field
 46    flags for the reverse relation.
 47    """
 48
 49    # Field flags
 50    auto_created = True
 51    concrete = False
 52
 53    # Reverse relations are always nullable (Plain can't enforce that a
 54    # foreign key on the related model points to this model).
 55    allow_null = True
 56    empty_strings_allowed = False
 57
 58    # Type annotations for instance attributes
 59    model: type[Model]
 60    field: RelatedField
 61    on_delete: OnDeleteCallback | None
 62    limit_choices_to: dict[str, Any] | Q
 63
 64    def __init__(
 65        self,
 66        field: RelatedField,
 67        to: str | type[Model],
 68        related_query_name: str | None = None,
 69        limit_choices_to: dict[str, Any] | Q | None = None,
 70        on_delete: OnDeleteCallback | None = None,
 71    ):
 72        self.field = field
 73        # Initially may be a string, gets resolved to type[Model] by lazy_related_operation
 74        # (see related.py:250 where field.remote_field.model is overwritten)
 75        self.model = to  # type: ignore[assignment]
 76        self.related_query_name = related_query_name
 77        self.limit_choices_to = {} if limit_choices_to is None else limit_choices_to
 78        self.on_delete = on_delete
 79
 80        self.symmetrical = False
 81        self.multiple = True
 82
 83    # Some of the following cached_properties can't be initialized in
 84    # __init__ as the field doesn't have its model yet. Calling these methods
 85    # before field.contribute_to_class() has been called will result in
 86    # AttributeError
 87    @cached_property
 88    def name(self) -> str:
 89        return self.field.related_query_name()
 90
 91    @property
 92    def remote_field(self) -> RelatedField:
 93        return self.field
 94
 95    @property
 96    def target_field(self) -> Field:
 97        """
 98        When filtering against this relation, return the field on the remote
 99        model against which the filtering should happen.
100        """
101        target_fields = self.path_infos[-1].target_fields
102        if len(target_fields) > 1:
103            raise FieldError("Can't use target_field for multicolumn relations.")
104        return target_fields[0]
105
106    @cached_property
107    def related_model(self) -> type[Model]:
108        if not self.field.model:
109            raise AttributeError(
110                "This property can't be accessed before self.field.contribute_to_class "
111                "has been called."
112            )
113        return self.field.model
114
115    def get_lookup(self, lookup_name: str) -> type[Lookup] | None:
116        return self.field.get_lookup(lookup_name)
117
118    def get_internal_type(self) -> str:
119        return self.field.get_internal_type()
120
121    @property
122    def db_type(self) -> str | None:
123        return self.field.db_type
124
125    def __repr__(self) -> str:
126        return f"<{type(self).__name__}: {self.related_model.model_options.package_label}.{self.related_model.model_options.model_name}>"
127
128    @property
129    def identity(self) -> tuple[Any, ...]:
130        return (
131            self.field,
132            self.model,
133            self.related_query_name,
134            make_hashable(self.limit_choices_to),
135            self.on_delete,
136            self.symmetrical,
137            self.multiple,
138        )
139
140    def __eq__(self, other: object) -> bool:
141        if not isinstance(other, self.__class__):
142            return NotImplemented
143        return self.identity == other.identity
144
145    def __hash__(self) -> int:
146        return hash(self.identity)
147
148    def __getstate__(self) -> dict[str, Any]:
149        state = self.__dict__.copy()
150        # Delete the path_infos cached property because it can be recalculated
151        # at first invocation after deserialization. The attribute must be
152        # removed because subclasses like ForeignKeyRel may have a PathInfo
153        # which contains an intermediate M2M table that's been dynamically
154        # created and doesn't exist in the .models module.
155        # This is a reverse relation, so there is no reverse_path_infos to
156        # delete.
157        state.pop("path_infos", None)
158        return state
159
160    def get_choices(
161        self,
162        include_blank: bool = True,
163        blank_choice: list[tuple[str, str]] = BLANK_CHOICE_DASH,
164        limit_choices_to: Any = None,
165        ordering: tuple[str, ...] = (),
166    ) -> list[tuple[Any, str]]:
167        """
168        Return choices with a default blank choices included, for use
169        as <select> choices for this field.
170
171        Analog of plain.models.fields.Field.get_choices(), provided
172        initially for utilization by RelatedFieldListFilter.
173        """
174        limit_choices_to = limit_choices_to or self.limit_choices_to
175        qs = self.related_model.query.complex_filter(limit_choices_to)
176        if ordering:
177            qs = qs.order_by(*ordering)
178        return (blank_choice if include_blank else []) + [(x.id, str(x)) for x in qs]
179
180    def get_joining_columns(self) -> tuple[tuple[str, str], ...]:
181        return self.field.get_reverse_joining_columns()
182
183    def set_field_name(self) -> None:
184        """
185        Set the related field's name, this is not available until later stages
186        of app loading, so set_field_name is called from
187        set_attributes_from_rel()
188        """
189        # By default foreign object doesn't relate to any remote field (for
190        # example custom multicolumn joins currently have no remote field).
191        self.field_name = None
192
193    def get_path_info(self, filtered_relation: Any = None) -> list[PathInfo]:
194        if filtered_relation:
195            return self.field.get_reverse_path_info(filtered_relation)
196        else:
197            return self.field.reverse_path_infos
198
199    @cached_property
200    def path_infos(self) -> list[PathInfo]:
201        return self.get_path_info()
202
203    def get_cache_name(self) -> str | None:
204        """
205        Return the name of the cache key to use for storing an instance of the
206        forward model on the reverse model.
207
208        Uses the related_query_name for caching, which provides a stable name
209        for prefetch_related operations.
210        """
211        return self.field.related_query_name()
212
213
214class ForeignKeyRel(ForeignObjectRel):
215    """
216    Used by the ForeignKeyField field to store information about the relation.
217
218    ``_model_meta.get_fields()`` returns this class to provide access to the
219    reverse relation. Use ``isinstance(rel, ForeignKeyRel)`` to identify
220    one-to-many reverse relations.
221    """
222
223    # Type annotations for instance attributes
224    field: ForeignKeyField
225
226    def __init__(
227        self,
228        field: ForeignKeyField,
229        to: str | type[Model],
230        related_query_name: str | None = None,
231        limit_choices_to: dict[str, Any] | Q | None = None,
232        on_delete: OnDeleteCallback | None = None,
233    ):
234        super().__init__(
235            field,
236            to,
237            related_query_name=related_query_name,
238            limit_choices_to=limit_choices_to,
239            on_delete=on_delete,
240        )
241
242        self.field_name = "id"
243
244    def __getstate__(self) -> dict[str, Any]:
245        state = super().__getstate__()
246        state.pop("related_model", None)
247        return state
248
249    @property
250    def identity(self) -> tuple[Any, ...]:
251        return super().identity + (self.field_name,)
252
253    def get_related_field(self) -> Field:
254        """
255        Return the Field in the 'to' object to which this relationship is tied.
256        """
257        return self.model._model_meta.get_forward_field("id")
258
259    def set_field_name(self) -> None:
260        pass
261
262
263class ManyToManyRel(ForeignObjectRel):
264    """
265    Used by ManyToManyField to store information about the relation.
266
267    ``_model_meta.get_fields()`` returns this class to provide access to the field
268    flags for the reverse relation.
269    """
270
271    # Type annotations for instance attributes
272    field: ManyToManyField
273    through: type[Model]
274    through_fields: tuple[str, str] | None
275
276    def __init__(
277        self,
278        field: ManyToManyField,
279        to: str | type[Model],
280        *,
281        through: str | type[Model],
282        through_fields: tuple[str, str] | None = None,
283        related_query_name: str | None = None,
284        limit_choices_to: dict[str, Any] | Q | None = None,
285        symmetrical: bool = True,
286    ):
287        super().__init__(
288            field,
289            to,
290            related_query_name=related_query_name,
291            limit_choices_to=limit_choices_to,
292        )
293
294        # Initially may be a string, gets resolved to type[Model] by lazy_related_operation
295        # (see related.py:1143 where field.remote_field.through is overwritten)
296        self.through = through  # type: ignore[assignment]
297        self.through_fields = through_fields
298
299        self.symmetrical = symmetrical
300        self.db_constraint = True
301
302    @property
303    def identity(self) -> tuple[Any, ...]:
304        return super().identity + (
305            self.through,
306            make_hashable(self.through_fields),
307            self.db_constraint,
308        )
309
310    def get_related_field(self) -> Field:
311        """
312        Return the field in the 'to' object to which this relationship is tied.
313        Provided for symmetry with ForeignKeyRel.
314        """
315        from plain.models.fields.related import ForeignKeyField
316
317        meta = self.through._model_meta
318        if self.through_fields:
319            field = meta.get_forward_field(self.through_fields[0])
320        else:
321            for field in meta.fields:
322                rel = getattr(field, "remote_field", None)
323                if rel and rel.model == self.model:
324                    break
325
326        if not isinstance(field, ForeignKeyField):
327            raise ValueError(f"Expected ForeignKeyField, got {type(field)}")
328        return field.foreign_related_fields[0]