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: ReverseForeignKey[Child] = ReverseForeignKey(to="Child", field="parent")
16
17 class Child(Model):
18 parent: Parent = ForeignKeyField(Parent, on_delete=models.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
26 Uniqueness of foreign key values is irrelevant to accessing the related
27 instance, making the many-to-one and one-to-one cases identical as far as
28 the descriptor is concerned. The constraint is checked upstream (unicity
29 validation in forms) or downstream (unique indexes in the database).
30
312. Related objects manager for related instances on the forward or reverse
32 sides of a many-to-many relation: ``ForwardManyToManyDescriptor``.
33
34 Many-to-many relations are symmetrical. The syntax of Plain models
35 requires declaring them on one side but that's an implementation detail.
36 They could be declared on the other side without any change in behavior.
37
38Reverse relations must be explicitly defined using ``ReverseForeignKey`` or
39``ReverseManyToMany`` descriptors on the model class.
40"""
41
42from __future__ import annotations
43
44from functools import cached_property
45from typing import Any
46
47from plain.models.query import QuerySet
48from plain.utils.functional import LazyObject
49
50from .related_managers import ManyToManyManager
51
52
53class ForwardForeignKeyDescriptor:
54 """
55 Accessor to the related object on the forward side of a foreign key relation.
56
57 In the example::
58
59 class Child(Model):
60 parent: Parent = ForeignKeyField(Parent, on_delete=models.CASCADE)
61
62 ``Child.parent`` is a ``ForwardForeignKeyDescriptor`` instance.
63 """
64
65 def __init__(self, field_with_rel: Any) -> None:
66 self.field = field_with_rel
67
68 @cached_property
69 def RelatedObjectDoesNotExist(self) -> type:
70 # The exception can't be created at initialization time since the
71 # related model might not be resolved yet; `self.field.model` might
72 # still be a string model reference.
73 return type(
74 "RelatedObjectDoesNotExist",
75 (self.field.remote_field.model.DoesNotExist, AttributeError),
76 {
77 "__module__": self.field.model.__module__,
78 "__qualname__": f"{self.field.model.__qualname__}.{self.field.name}.RelatedObjectDoesNotExist",
79 },
80 )
81
82 def is_cached(self, instance: Any) -> bool:
83 return self.field.is_cached(instance)
84
85 def get_queryset(self) -> QuerySet:
86 qs = self.field.remote_field.model._model_meta.base_queryset
87 return qs.all()
88
89 def get_prefetch_queryset(
90 self, instances: list[Any], queryset: QuerySet | None = None
91 ) -> tuple[QuerySet, Any, Any, bool, str, bool]:
92 if queryset is None:
93 queryset = self.get_queryset()
94
95 rel_obj_attr = self.field.get_foreign_related_value
96 instance_attr = self.field.get_local_related_value
97 instances_dict = {instance_attr(inst): inst for inst in instances}
98 related_field = self.field.foreign_related_fields[0]
99 remote_field = self.field.remote_field
100
101 # FIXME: This will need to be revisited when we introduce support for
102 # composite fields. In the meantime we take this practical approach.
103 # Refs #21410.
104 # The check for len(...) == 1 is a special case that allows the query
105 # to be join-less and smaller. Refs #21760.
106 if len(self.field.foreign_related_fields) == 1:
107 query = {
108 f"{related_field.name}__in": {
109 instance_attr(inst)[0] for inst in instances
110 }
111 }
112 else:
113 query = {f"{self.field.related_query_name()}__in": instances}
114 queryset = queryset.filter(**query)
115
116 # Since we're going to assign directly in the cache,
117 # we must manage the reverse relation cache manually.
118 if not remote_field.multiple:
119 for rel_obj in queryset:
120 instance = instances_dict[rel_obj_attr(rel_obj)]
121 remote_field.set_cached_value(rel_obj, instance)
122 return (
123 queryset,
124 rel_obj_attr,
125 instance_attr,
126 True,
127 self.field.get_cache_name(),
128 False,
129 )
130
131 def get_object(self, instance: Any) -> Any:
132 qs = self.get_queryset()
133 # Assuming the database enforces foreign keys, this won't fail.
134 return qs.get(self.field.get_reverse_related_filter(instance))
135
136 def __get__(
137 self, instance: Any | None, cls: type | None = None
138 ) -> ForwardForeignKeyDescriptor | Any | None:
139 """
140 Get the related instance through the forward relation.
141
142 With the example above, when getting ``child.parent``:
143
144 - ``self`` is the descriptor managing the ``parent`` attribute
145 - ``instance`` is the ``child`` instance
146 - ``cls`` is the ``Child`` class (we don't need it)
147 """
148 if instance is None:
149 return self
150
151 # The related instance is loaded from the database and then cached
152 # by the field on the model instance state. It can also be pre-cached
153 # by the reverse accessor.
154 try:
155 rel_obj = self.field.get_cached_value(instance)
156 except KeyError:
157 has_value = None not in self.field.get_local_related_value(instance)
158 rel_obj = None
159
160 if rel_obj is None and has_value:
161 rel_obj = self.get_object(instance)
162 remote_field = self.field.remote_field
163 # If this is a one-to-one relation, set the reverse accessor
164 # cache on the related object to the current instance to avoid
165 # an extra SQL query if it's accessed later on.
166 if not remote_field.multiple:
167 remote_field.set_cached_value(rel_obj, instance)
168 self.field.set_cached_value(instance, rel_obj)
169
170 if rel_obj is None and not self.field.allow_null:
171 raise self.RelatedObjectDoesNotExist(
172 f"{self.field.model.__name__} has no {self.field.name}."
173 )
174 else:
175 return rel_obj
176
177 def __set__(self, instance: Any, value: Any) -> None:
178 """
179 Set the related instance through the forward relation.
180
181 With the example above, when setting ``child.parent = parent``:
182
183 - ``self`` is the descriptor managing the ``parent`` attribute
184 - ``instance`` is the ``child`` instance
185 - ``value`` is the ``parent`` instance on the right of the equal sign
186 """
187 # If value is a LazyObject, force its evaluation. For ForeignKeyField fields,
188 # the value should only be None or a model instance, never a boolean or
189 # other type.
190 if isinstance(value, LazyObject):
191 # This forces evaluation: if it's None, value becomes None;
192 # if it's a User instance, value becomes that instance.
193 value = value if value else None
194
195 # An object must be an instance of the related class.
196 if value is not None and not isinstance(value, self.field.remote_field.model):
197 raise ValueError(
198 f'Cannot assign "{value!r}": "{instance.model_options.object_name}.{self.field.name}" must be a "{self.field.remote_field.model.model_options.object_name}" instance.'
199 )
200 remote_field = self.field.remote_field
201 # If we're setting the value of a OneToOneField to None, we need to clear
202 # out the cache on any old related object. Otherwise, deleting the
203 # previously-related object will also cause this object to be deleted,
204 # which is wrong.
205 if value is None:
206 # Look up the previously-related object, which may still be available
207 # since we've not yet cleared out the related field.
208 # Use the cache directly, instead of the accessor; if we haven't
209 # populated the cache, then we don't care - we're only accessing
210 # the object to invalidate the accessor cache, so there's no
211 # need to populate the cache just to expire it again.
212 related = self.field.get_cached_value(instance, default=None)
213
214 # If we've got an old related object, we need to clear out its
215 # cache. This cache also might not exist if the related object
216 # hasn't been accessed yet.
217 if related is not None:
218 remote_field.set_cached_value(related, None)
219
220 for lh_field, rh_field in self.field.related_fields:
221 setattr(instance, lh_field.attname, None)
222
223 # Set the values of the related field.
224 else:
225 for lh_field, rh_field in self.field.related_fields:
226 setattr(instance, lh_field.attname, getattr(value, rh_field.attname))
227
228 # Set the related instance cache used by __get__ to avoid an SQL query
229 # when accessing the attribute we just set.
230 self.field.set_cached_value(instance, value)
231
232 # If this is a one-to-one relation, set the reverse accessor cache on
233 # the related object to the current instance to avoid an extra SQL
234 # query if it's accessed later on.
235 if value is not None and not remote_field.multiple:
236 remote_field.set_cached_value(value, instance)
237
238 def __reduce__(self) -> tuple[Any, tuple[Any, str]]:
239 """
240 Pickling should return the instance attached by self.field on the
241 model, not a new copy of that descriptor. Use getattr() to retrieve
242 the instance directly from the model.
243 """
244 return getattr, (self.field.model, self.field.name)
245
246
247class ForwardManyToManyDescriptor:
248 """
249 Accessor to the related objects manager on the forward side of a
250 many-to-many relation.
251
252 In the example::
253
254 class Pizza(Model):
255 toppings: ManyToManyField[Topping] = ManyToManyField(Topping, through=PizzaTopping)
256
257 ``Pizza.toppings`` is a ``ForwardManyToManyDescriptor`` instance.
258 """
259
260 def __init__(self, rel: Any) -> None:
261 self.rel = rel
262 self.field = rel.field
263
264 def __get__(
265 self, instance: Any | None, cls: type | None = None
266 ) -> ForwardManyToManyDescriptor | Any:
267 """Get the related manager when the descriptor is accessed."""
268 if instance is None:
269 return self
270 return ManyToManyManager(
271 instance=instance,
272 field=self.rel.field,
273 through=self.rel.through,
274 related_model=self.rel.model,
275 is_reverse=False,
276 symmetrical=self.rel.symmetrical,
277 )
278
279 def __set__(self, instance: Any, value: Any) -> None:
280 """Prevent direct assignment to the relation."""
281 raise TypeError(
282 f"Direct assignment to the forward side of a many-to-many set is prohibited. Use {self.field.name}.set() instead.",
283 )
284
285 @property
286 def through(self) -> Any:
287 # through is provided so that you have easy access to the through
288 # model (Book.authors.through) for inlines, etc. This is done as
289 # a property to ensure that the fully resolved value is returned.
290 return self.rel.through