1"""
2Accessors for related objects.
3
4When a field defines a relation between two models, the forward model provides
5an attribute to access related instances. Reverse accessors must be explicitly
6defined using ReverseForeignKey or ReverseManyToMany descriptors.
7
8Accessors are implemented as descriptors in order to customize access and
9assignment. This module defines the descriptor classes.
10
11Forward accessors follow foreign keys. Reverse accessors trace them back. For
12example, with the following models::
13
14 class Parent(Model):
15 children: ClassVar[ReverseForeignKey[Child]] = ReverseForeignKey(to="Child", field="parent")
16
17 class Child(Model):
18 parent: Field[Parent] = ForeignKeyField(Parent, on_delete=CASCADE)
19
20 ``child.parent`` is a forward foreign key relation. ``parent.children`` is a
21reverse foreign key relation.
22
231. Related instance on the forward side of a foreign key relation:
24 ``ForwardForeignKeyDescriptor``.
25
262. Related objects manager for related instances on the forward or reverse
27 sides of a many-to-many relation: ``ForwardManyToManyDescriptor``.
28
29 Many-to-many relations are symmetrical. The syntax of Plain models
30 requires declaring them on one side but that's an implementation detail.
31 They could be declared on the other side without any change in behavior.
32
33Reverse relations must be explicitly defined using ``ReverseForeignKey`` or
34``ReverseManyToMany`` descriptors on the model class.
35"""
36
37from functools import cached_property
38from typing import Any
39
40from plain.postgres.query import QuerySet
41from plain.utils.functional import LazyObject
42
43from .related_managers import ManyToManyManager
44
45
46class ForwardForeignKeyDescriptor:
47 """
48 Accessor to the related object on the forward side of a foreign key relation.
49
50 In the example::
51
52 class Child(Model):
53 parent: Field[Parent] = ForeignKeyField(Parent, on_delete=postgres.CASCADE)
54
55 ``Child.parent`` is a ``ForwardForeignKeyDescriptor`` instance.
56 """
57
58 def __init__(self, field_with_rel: Any) -> None:
59 self._field = field_with_rel
60
61 @cached_property
62 def RelatedObjectDoesNotExist(self) -> type:
63 # The exception can't be created at initialization time since the
64 # related model might not be resolved yet; `self._field.model` might
65 # still be a string model reference.
66 return type(
67 "RelatedObjectDoesNotExist",
68 (self._field.remote_field.model.DoesNotExist, AttributeError),
69 {
70 "__module__": self._field.model.__module__,
71 "__qualname__": f"{self._field.model.__qualname__}.{self._field.name}.RelatedObjectDoesNotExist",
72 },
73 )
74
75 def _is_cached(self, instance: Any) -> bool:
76 return self._field.is_cached(instance)
77
78 def _get_queryset(self) -> QuerySet:
79 qs = self._field.remote_field.model._model_meta.base_queryset
80 return qs.all()
81
82 def _get_prefetch_queryset(
83 self, instances: list[Any], queryset: QuerySet | None = None
84 ) -> tuple[QuerySet, Any, Any, bool, str, bool]:
85 if queryset is None:
86 queryset = self._get_queryset()
87
88 rel_obj_attr = self._field.get_foreign_related_value
89 instance_attr = self._field.get_local_related_value
90 related_field = self._field.target_field
91
92 # A foreign key is single-column, so prefetch with a join-less IN query.
93 query = {
94 f"{related_field.name}__in": {instance_attr(inst) for inst in instances}
95 }
96 queryset = queryset.filter(**query)
97
98 return (
99 queryset,
100 rel_obj_attr,
101 instance_attr,
102 True,
103 self._field.get_cache_name(),
104 False,
105 )
106
107 def __get__(self, instance: Any | None, cls: type | None = None) -> Any:
108 """
109 Get the related instance through the forward relation.
110
111 With the example above, when getting ``child.parent``:
112
113 - ``self`` is the descriptor managing the ``parent`` attribute
114 - ``instance`` is the ``child`` instance
115 - ``cls`` is the ``Child`` class (we don't need it)
116
117 Class-level access (``Child.parent``) returns the descriptor itself,
118 the ordinary convention. Traversal for typed where() is served by
119 ``__getattr__`` below rather than by handing back a proxy.
120 """
121 if instance is None:
122 return self
123
124 # The related object is cached on the model state -- by join(),
125 # prefetch, the reverse accessor, a prior access, or assignment.
126 try:
127 rel_obj = self._field.get_cached_value(instance)
128 except KeyError:
129 # _get_raw_value loads the foreign key column on demand if it was
130 # deferred (.only()/.defer()), so we always see the real key here.
131 pk_value = self._field._get_raw_value(instance)
132 rel_obj = None
133 if pk_value is not None:
134 remote_model = self._field.remote_field.model
135 target_name = self._field.target_field.name
136 assert target_name is not None
137 # The database FK constraint guarantees the row exists, so build
138 # a partial related instance with only its primary key loaded --
139 # no query. Accessing any other field triggers the full-row
140 # deferred load.
141 rel_obj = remote_model.from_db([target_name], [pk_value])
142 self._field.set_cached_value(instance, rel_obj)
143
144 # Checked on every access, including a cached None: a non-nullable
145 # foreign key with no value must raise consistently, not just once.
146 if rel_obj is None and not self._field.allow_null:
147 raise self.RelatedObjectDoesNotExist(
148 f"{self._field.model.__name__} has no {self._field.name}."
149 )
150 return rel_obj
151
152 def _build_typed_ref(self) -> Any:
153 """The traversal entry point for this relation.
154
155 Built lazily, and cached by `__getattr__` rather than by
156 `cached_property`: this can raise `UnresolvedRelationError`, and an
157 AttributeError raised inside a descriptor's `__get__` makes Python
158 fall back to `__getattr__`, which would replace the useful message
159 with a bare "no attribute '_typed_ref'".
160
161 The import is local because `related_typed` reaches `fields.related`,
162 which imports this module at load time.
163 """
164 from plain.postgres.fields.related_typed import (
165 RelatedFieldRef,
166 unresolved_relation_error,
167 )
168
169 try:
170 model = self._field.remote_field.model
171 except TypeError:
172 # `ForeignObjectRel.model` raises TypeError while the target is
173 # still a string, which would escape __getattr__ as a TypeError and
174 # make `hasattr` raise instead of returning False.
175 raise unresolved_relation_error(
176 self._field.name, str(self._field.remote_field.model_ref)
177 ) from None
178
179 return RelatedFieldRef(
180 model=model,
181 prefix=self._field.name,
182 target_name=self._field.target_field.name,
183 # The traversal starts here, on the model the relation is declared
184 # on -- that is the model a condition built through it belongs to.
185 source_model=self._field.model,
186 )
187
188 def __getattr__(self, name: str) -> Any:
189 """Walk class-level attribute access into the related model, so typed
190 where() can build joined lookups:
191
192 Child.parent.name.equals("x") → Q(parent__name="x")
193
194 Only names this descriptor doesn't define reach here, and every
195 attribute it does define is ``_``-prefixed -- which model fields can
196 never be (``Meta`` skips ``_``-prefixed attributes when it collects
197 them), so a related field can't be shadowed by descriptor internals.
198
199 Delegated to `RelatedFieldRef` so the first hop follows exactly the
200 same rule as every later one.
201 """
202 if name.startswith("_"):
203 # Internals, and anything a field could never be named.
204 raise AttributeError(name)
205 ref = self.__dict__.get("_typed_ref")
206 if ref is None:
207 ref = self.__dict__["_typed_ref"] = self._build_typed_ref()
208 return getattr(ref, name)
209
210 def __set__(self, instance: Any, value: Any) -> None:
211 """
212 Set the related object (or its raw key) through the forward relation.
213
214 Accepts a related model instance, a bare primary key value, or None::
215
216 child.parent = parent_instance
217 child.parent = 5
218 child.parent = None
219 """
220 from plain.postgres.base import Model
221
222 # A LazyObject (e.g. request.user) must be evaluated before use.
223 if isinstance(value, LazyObject):
224 value = value if value else None
225
226 name = self._field.name
227 assert name is not None
228 remote_field = self._field.remote_field
229
230 if value is None:
231 instance.__dict__[name] = None
232 self._field.set_cached_value(instance, None)
233 return
234
235 if isinstance(value, remote_field.model):
236 # A related model instance: store its key, cache the object.
237 instance.__dict__[name] = getattr(value, self._field.target_field.name)
238 self._field.set_cached_value(instance, value)
239 return
240
241 if isinstance(value, Model | bool):
242 # A wrong-model instance, or a bool (which would silently coerce to
243 # the key 0/1 via int) -- reject rather than store a bogus key.
244 raise TypeError(
245 f'Cannot assign "{value!r}": '
246 f'"{instance.model_options.object_name}.{self._field.name}" must be a '
247 f'"{remote_field.model.model_options.object_name}" instance or a '
248 f"primary key value."
249 )
250
251 # A bare related key value (e.g. child.parent = 5).
252 new_value = self._field.to_python(value)
253 # On an actual key change, drop the now-stale forward cache.
254 # Re-storing the same key (e.g. by clean_fields) keeps the cache.
255 if instance.__dict__.get(name) != new_value and self._field.is_cached(instance):
256 self._field.delete_cached_value(instance)
257 instance.__dict__[name] = new_value
258
259 def __delete__(self, instance: Any) -> None:
260 """Delete the foreign key value, clearing any cached related object."""
261 try:
262 del instance.__dict__[self._field.name]
263 except KeyError:
264 raise AttributeError(
265 f"{instance.__class__.__name__!r} object has no attribute "
266 f"{self._field.name!r}"
267 )
268 if self._field.is_cached(instance):
269 self._field.delete_cached_value(instance)
270
271 def __reduce__(self) -> tuple[Any, tuple[Any, str]]:
272 """
273 Pickling should return the instance attached by self._field on the
274 model, not a new copy of that descriptor.
275
276 Class access returns the descriptor, so ``getattr`` retrieves the
277 instance directly from the model.
278 """
279 return getattr, (self._field.model, self._field.name)
280
281
282class ForwardManyToManyDescriptor:
283 """
284 Accessor to the related objects manager on the forward side of a
285 many-to-many relation.
286
287 In the example::
288
289 class Pizza(Model):
290 toppings: ManyToManyField[Topping] = ManyToManyField(Topping, through=PizzaTopping)
291
292 ``Pizza.toppings`` is a ``ForwardManyToManyDescriptor`` instance.
293 """
294
295 def __init__(self, rel: Any) -> None:
296 self.rel = rel
297 self.field = rel.field
298
299 def __get__(
300 self, instance: Any | None, cls: type | None = None
301 ) -> ForwardManyToManyDescriptor | Any:
302 """Get the related manager when the descriptor is accessed."""
303 if instance is None:
304 return self
305 return ManyToManyManager(
306 instance=instance,
307 field=self.rel.field,
308 through=self.rel.through,
309 related_model=self.rel.model,
310 is_reverse=False,
311 symmetrical=self.rel.symmetrical,
312 )
313
314 def __set__(self, instance: Any, value: Any) -> None:
315 """Prevent direct assignment to the relation."""
316 raise TypeError(
317 f"Direct assignment to the forward side of a many-to-many set is prohibited. Use {self.field.name}.set() instead.",
318 )
319
320 @property
321 def through(self) -> Any:
322 # through is provided so that you have easy access to the through
323 # model (Book.authors.through) for inlines, etc. This is done as
324 # a property to ensure that the fully resolved value is returned.
325 return self.rel.through