Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/mptt/managers.py: 27%
530 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1"""
2A custom manager for working with trees of objects.
3"""
4import contextlib
5import functools
6from itertools import groupby
8from django.db import connections, models, router
9from django.db.models import (
10 F,
11 IntegerField,
12 ManyToManyField,
13 Max,
14 OuterRef,
15 Q,
16 Subquery,
17)
18from django.utils.translation import gettext as _
20from mptt.compat import cached_field_value
21from mptt.exceptions import CantDisableUpdates, InvalidMove
22from mptt.querysets import TreeQuerySet
23from mptt.signals import node_moved
24from mptt.utils import _get_tree_model
27__all__ = ("TreeManager",)
30class SQCount(Subquery):
31 template = "(SELECT count(*) FROM (%(subquery)s) _count)"
32 output_field = IntegerField()
35def delegate_manager(method):
36 """
37 Delegate method calls to base manager, if exists.
38 """
40 @functools.wraps(method)
41 def wrapped(self, *args, **kwargs):
42 if self._base_manager: 42 ↛ 43line 42 didn't jump to line 43, because the condition on line 42 was never true
43 return getattr(self._base_manager, method.__name__)(*args, **kwargs)
44 return method(self, *args, **kwargs)
46 return wrapped
49class TreeManager(models.Manager.from_queryset(TreeQuerySet)):
51 """
52 A manager for working with trees of objects.
53 """
55 def contribute_to_class(self, model, name):
56 super().contribute_to_class(model, name)
58 if not model._meta.abstract:
59 self.tree_model = _get_tree_model(model)
61 self._base_manager = None
62 if self.tree_model and self.tree_model is not model: 62 ↛ 64line 62 didn't jump to line 64, because the condition on line 62 was never true
63 # _base_manager is the treemanager on tree_model
64 self._base_manager = self.tree_model._tree_manager
66 def get_queryset(self, *args, **kwargs):
67 """
68 Ensures that this manager always returns nodes in tree order.
69 """
70 return (
71 super()
72 .get_queryset(*args, **kwargs)
73 .order_by(self.tree_id_attr, self.left_attr)
74 )
76 def _get_queryset_relatives(self, queryset, direction, include_self):
77 """
78 Returns a queryset containing either the descendants
79 ``direction == desc`` or the ancestors ``direction == asc`` of a given
80 queryset.
82 This function is not meant to be called directly, although there is no
83 harm in doing so.
85 Instead, it should be used via ``get_queryset_descendants()`` and/or
86 ``get_queryset_ancestors()``.
88 This function works by grouping contiguous siblings and using them to create
89 a range that selects all nodes between the range, instead of querying for each
90 node individually. Three variables are required when querying for ancestors or
91 descendants: tree_id_attr, left_attr, right_attr. If we weren't using ranges
92 and our queryset contained 100 results, the resulting SQL query would contain
93 300 variables. However, when using ranges, if the same queryset contained 10
94 sets of contiguous siblings, then the resulting SQL query should only contain
95 30 variables.
97 The attributes used to create the range are completely
98 dependent upon whether you are ascending or descending the tree.
100 * Ascending (ancestor nodes): select all nodes whose right_attr is greater
101 than (or equal to, if include_self = True) the smallest right_attr within
102 the set of contiguous siblings, and whose left_attr is less than (or equal
103 to) the largest left_attr within the set of contiguous siblings.
105 * Descending (descendant nodes): select all nodes whose left_attr is greater
106 than (or equal to, if include_self = True) the smallest left_attr within
107 the set of contiguous siblings, and whose right_attr is less than (or equal
108 to) the largest right_attr within the set of contiguous siblings.
110 The result is the more contiguous siblings in the original queryset, the fewer
111 SQL variables will be required to execute the query.
112 """
113 assert self.model is queryset.model
115 opts = queryset.model._mptt_meta
117 filters = Q()
119 e = "e" if include_self else ""
120 max_op = "lt" + e
121 min_op = "gt" + e
122 if direction == "asc":
123 max_attr = opts.left_attr
124 min_attr = opts.right_attr
125 elif direction == "desc":
126 max_attr = opts.right_attr
127 min_attr = opts.left_attr
129 tree_key = opts.tree_id_attr
130 min_key = "%s__%s" % (min_attr, min_op)
131 max_key = "%s__%s" % (max_attr, max_op)
133 q = queryset.order_by(opts.tree_id_attr, opts.parent_attr, opts.left_attr).only(
134 opts.tree_id_attr,
135 opts.left_attr,
136 opts.right_attr,
137 min_attr,
138 max_attr,
139 opts.parent_attr,
140 # These fields are used by MPTTModel.update_mptt_cached_fields()
141 *[f.lstrip("-") for f in opts.order_insertion_by]
142 )
144 if not q:
145 return self.none()
147 for group in groupby(
148 q,
149 key=lambda n: (
150 getattr(n, opts.tree_id_attr),
151 getattr(n, opts.parent_attr + "_id"),
152 ),
153 ):
154 next_lft = None
155 for node in list(group[1]):
156 tree, lft, rght, min_val, max_val = (
157 getattr(node, opts.tree_id_attr),
158 getattr(node, opts.left_attr),
159 getattr(node, opts.right_attr),
160 getattr(node, min_attr),
161 getattr(node, max_attr),
162 )
163 if next_lft is None:
164 next_lft = rght + 1
165 min_max = {"min": min_val, "max": max_val}
166 elif lft == next_lft:
167 if min_val < min_max["min"]:
168 min_max["min"] = min_val
169 if max_val > min_max["max"]:
170 min_max["max"] = max_val
171 next_lft = rght + 1
172 elif lft != next_lft:
173 filters |= Q(
174 **{
175 tree_key: tree,
176 min_key: min_max["min"],
177 max_key: min_max["max"],
178 }
179 )
180 min_max = {"min": min_val, "max": max_val}
181 next_lft = rght + 1
182 filters |= Q(
183 **{
184 tree_key: tree,
185 min_key: min_max["min"],
186 max_key: min_max["max"],
187 }
188 )
190 return self.filter(filters)
192 def get_queryset_descendants(self, queryset, include_self=False):
193 """
194 Returns a queryset containing the descendants of all nodes in the
195 given queryset.
197 If ``include_self=True``, nodes in ``queryset`` will also
198 be included in the result.
199 """
200 return self._get_queryset_relatives(queryset, "desc", include_self)
202 def get_queryset_ancestors(self, queryset, include_self=False):
203 """
204 Returns a queryset containing the ancestors
205 of all nodes in the given queryset.
207 If ``include_self=True``, nodes in ``queryset`` will also
208 be included in the result.
209 """
210 return self._get_queryset_relatives(queryset, "asc", include_self)
212 @contextlib.contextmanager
213 def disable_mptt_updates(self):
214 """
215 Context manager. Disables mptt updates.
217 NOTE that this context manager causes inconsistencies! MPTT model
218 methods are not guaranteed to return the correct results.
220 When to use this method:
221 If used correctly, this method can be used to speed up bulk
222 updates.
224 This doesn't do anything clever. It *will* mess up your tree. You
225 should follow this method with a call to ``TreeManager.rebuild()``
226 to ensure your tree stays sane, and you should wrap both calls in a
227 transaction.
229 This is best for updates that span a large part of the table. If
230 you are doing localised changes (one tree, or a few trees) consider
231 using ``delay_mptt_updates``.
233 If you are making only minor changes to your tree, just let the
234 updates happen.
236 Transactions:
237 This doesn't enforce any transactional behavior. You should wrap
238 this in a transaction to ensure database consistency.
240 If updates are already disabled on the model, this is a noop.
242 Usage::
244 with transaction.atomic():
245 with MyNode.objects.disable_mptt_updates():
246 ## bulk updates.
247 MyNode.objects.rebuild()
248 """
249 # Error cases:
250 if self.model._meta.abstract:
251 # an abstract model. Design decision needed - do we disable
252 # updates for all concrete models that derive from this model? I
253 # vote no - that's a bit implicit and it's a weird use-case
254 # anyway. Open to further discussion :)
255 raise CantDisableUpdates(
256 "You can't disable/delay mptt updates on %s,"
257 " it's an abstract model" % self.model.__name__
258 )
259 elif self.model._meta.proxy:
260 # a proxy model. disabling updates would implicitly affect other
261 # models using the db table. Caller should call this on the
262 # manager for the concrete model instead, to make the behavior
263 # explicit.
264 raise CantDisableUpdates(
265 "You can't disable/delay mptt updates on %s, it's a proxy"
266 " model. Call the concrete model instead." % self.model.__name__
267 )
268 elif self.tree_model is not self.model:
269 # a multiple-inheritance child of an MPTTModel. Disabling
270 # updates may affect instances of other models in the tree.
271 raise CantDisableUpdates(
272 "You can't disable/delay mptt updates on %s, it doesn't"
273 " contain the mptt fields." % self.model.__name__
274 )
276 if not self.model._mptt_updates_enabled:
277 # already disabled, noop.
278 yield
279 else:
280 self.model._set_mptt_updates_enabled(False)
281 try:
282 yield
283 finally:
284 self.model._set_mptt_updates_enabled(True)
286 @contextlib.contextmanager
287 def delay_mptt_updates(self):
288 """
289 Context manager. Delays mptt updates until the end of a block of bulk
290 processing.
292 NOTE that this context manager causes inconsistencies! MPTT model
293 methods are not guaranteed to return the correct results until the end
294 of the context block.
296 When to use this method:
297 If used correctly, this method can be used to speed up bulk
298 updates. This is best for updates in a localised area of the db
299 table, especially if all the updates happen in a single tree and
300 the rest of the forest is left untouched. No subsequent rebuild is
301 necessary.
303 ``delay_mptt_updates`` does a partial rebuild of the modified trees
304 (not the whole table). If used indiscriminately, this can actually
305 be much slower than just letting the updates occur when they're
306 required.
308 The worst case occurs when every tree in the table is modified just
309 once. That results in a full rebuild of the table, which can be
310 *very* slow.
312 If your updates will modify most of the trees in the table (not a
313 small number of trees), you should consider using
314 ``TreeManager.disable_mptt_updates``, as it does much fewer
315 queries.
317 Transactions:
318 This doesn't enforce any transactional behavior. You should wrap
319 this in a transaction to ensure database consistency.
321 Exceptions:
322 If an exception occurs before the processing of the block, delayed
323 updates will not be applied.
325 Usage::
327 with transaction.atomic():
328 with MyNode.objects.delay_mptt_updates():
329 ## bulk updates.
330 """
331 with self.disable_mptt_updates():
332 if self.model._mptt_is_tracking:
333 # already tracking, noop.
334 yield
335 else:
336 self.model._mptt_start_tracking()
337 try:
338 yield
339 except Exception:
340 # stop tracking, but discard results
341 self.model._mptt_stop_tracking()
342 raise
343 results = self.model._mptt_stop_tracking()
344 partial_rebuild = self.partial_rebuild
345 for tree_id in results:
346 partial_rebuild(tree_id)
348 @property
349 def parent_attr(self):
350 return self.model._mptt_meta.parent_attr
352 @property
353 def left_attr(self):
354 return self.model._mptt_meta.left_attr
356 @property
357 def right_attr(self):
358 return self.model._mptt_meta.right_attr
360 @property
361 def tree_id_attr(self):
362 return self.model._mptt_meta.tree_id_attr
364 @property
365 def level_attr(self):
366 return self.model._mptt_meta.level_attr
368 def _translate_lookups(self, **lookups):
369 new_lookups = {}
370 join_parts = "__".join
371 for k, v in lookups.items():
372 parts = k.split("__")
373 new_parts = []
374 new_parts__append = new_parts.append
375 for part in parts:
376 new_parts__append(getattr(self, part + "_attr", part))
377 new_lookups[join_parts(new_parts)] = v
378 return new_lookups
380 @delegate_manager
381 def _mptt_filter(self, qs=None, **filters):
382 """
383 Like ``self.filter()``, but translates name-agnostic filters for MPTT
384 fields.
385 """
386 if qs is None: 386 ↛ 388line 386 didn't jump to line 388, because the condition on line 386 was never false
387 qs = self
388 return qs.filter(**self._translate_lookups(**filters))
390 @delegate_manager
391 def _mptt_update(self, qs=None, **items):
392 """
393 Like ``self.update()``, but translates name-agnostic MPTT fields.
394 """
395 if qs is None:
396 qs = self
397 return qs.update(**self._translate_lookups(**items))
399 def _get_connection(self, **hints):
400 return connections[router.db_for_write(self.model, **hints)]
402 def add_related_count(
403 self,
404 queryset,
405 rel_model,
406 rel_field,
407 count_attr,
408 cumulative=False,
409 extra_filters={},
410 ):
411 """
412 Adds a related item count to a given ``QuerySet`` using its
413 ``extra`` method, for a ``Model`` class which has a relation to
414 this ``Manager``'s ``Model`` class.
416 Arguments:
418 ``rel_model``
419 A ``Model`` class which has a relation to this `Manager``'s
420 ``Model`` class.
422 ``rel_field``
423 The name of the field in ``rel_model`` which holds the
424 relation.
426 ``count_attr``
427 The name of an attribute which should be added to each item in
428 this ``QuerySet``, containing a count of how many instances
429 of ``rel_model`` are related to it through ``rel_field``.
431 ``cumulative``
432 If ``True``, the count will be for each item and all of its
433 descendants, otherwise it will be for each item itself.
435 ``extra_filters``
436 Dict with aditional parameters filtering the related queryset.
437 """
438 if cumulative:
439 subquery_filters = {
440 rel_field + "__tree_id": OuterRef(self.tree_id_attr),
441 rel_field + "__lft__gte": OuterRef(self.left_attr),
442 rel_field + "__lft__lte": OuterRef(self.right_attr),
443 }
444 else:
445 current_rel_model = rel_model
446 for rel_field_part in rel_field.split("__"):
447 current_mptt_field = current_rel_model._meta.get_field(rel_field_part)
448 current_rel_model = current_mptt_field.related_model
449 mptt_field = current_mptt_field
451 if isinstance(mptt_field, ManyToManyField):
452 field_name = "pk"
453 else:
454 field_name = mptt_field.remote_field.field_name
456 subquery_filters = {
457 rel_field: OuterRef(field_name),
458 }
459 subquery = rel_model.objects.filter(**subquery_filters, **extra_filters).values(
460 "pk"
461 )
462 return queryset.annotate(**{count_attr: SQCount(subquery)})
464 @delegate_manager
465 def insert_node(
466 self,
467 node,
468 target,
469 position="last-child",
470 save=False,
471 allow_existing_pk=False,
472 refresh_target=True,
473 ):
474 """
475 Sets up the tree state for ``node`` (which has not yet been
476 inserted into in the database) so it will be positioned relative
477 to a given ``target`` node as specified by ``position`` (when
478 appropriate) it is inserted, with any necessary space already
479 having been made for it.
481 A ``target`` of ``None`` indicates that ``node`` should be
482 the last root node.
484 If ``save`` is ``True``, ``node``'s ``save()`` method will be
485 called before it is returned.
487 NOTE: This is a low-level method; it does NOT respect
488 ``MPTTMeta.order_insertion_by``. In most cases you should just
489 set the node's parent and let mptt call this during save.
490 """
492 if node.pk and not allow_existing_pk and self.filter(pk=node.pk).exists(): 492 ↛ 493line 492 didn't jump to line 493, because the condition on line 492 was never true
493 raise ValueError(_("Cannot insert a node which has already been saved."))
495 if target is None:
496 tree_id = self._get_next_tree_id()
497 setattr(node, self.left_attr, 1)
498 setattr(node, self.right_attr, 2)
499 setattr(node, self.level_attr, 0)
500 setattr(node, self.tree_id_attr, tree_id)
501 setattr(node, self.parent_attr, None)
502 elif target.is_root_node() and position in ["left", "right"]: 502 ↛ 503line 502 didn't jump to line 503, because the condition on line 502 was never true
503 if refresh_target:
504 # Ensure mptt values on target are not stale.
505 target._mptt_refresh()
507 target_tree_id = getattr(target, self.tree_id_attr)
508 if position == "left":
509 tree_id = target_tree_id
510 space_target = target_tree_id - 1
511 else:
512 tree_id = target_tree_id + 1
513 space_target = target_tree_id
514 self._create_tree_space(space_target)
516 setattr(node, self.left_attr, 1)
517 setattr(node, self.right_attr, 2)
518 setattr(node, self.level_attr, 0)
519 setattr(node, self.tree_id_attr, tree_id)
520 setattr(node, self.parent_attr, None)
521 else:
522 setattr(node, self.left_attr, 0)
523 setattr(node, self.level_attr, 0)
525 if refresh_target: 525 ↛ 529line 525 didn't jump to line 529
526 # Ensure mptt values on target are not stale.
527 target._mptt_refresh()
529 (
530 space_target,
531 level,
532 left,
533 parent,
534 right_shift,
535 ) = self._calculate_inter_tree_move_values(node, target, position)
537 tree_id = getattr(target, self.tree_id_attr)
538 self._create_space(2, space_target, tree_id)
540 setattr(node, self.left_attr, -left)
541 setattr(node, self.right_attr, -left + 1)
542 setattr(node, self.level_attr, -level)
543 setattr(node, self.tree_id_attr, tree_id)
544 setattr(node, self.parent_attr, parent)
546 if parent: 546 ↛ 549line 546 didn't jump to line 549, because the condition on line 546 was never false
547 self._post_insert_update_cached_parent_right(parent, right_shift)
549 if save: 549 ↛ 550line 549 didn't jump to line 550, because the condition on line 549 was never true
550 node.save()
551 return node
553 @delegate_manager
554 def _move_node(
555 self, node, target, position="last-child", save=True, refresh_target=True
556 ):
557 if self.tree_model._mptt_is_tracking:
558 # delegate to insert_node and clean up the gaps later.
559 return self.insert_node(
560 node,
561 target,
562 position=position,
563 save=save,
564 allow_existing_pk=True,
565 refresh_target=refresh_target,
566 )
567 else:
568 if target is None:
569 if node.is_child_node():
570 self._make_child_root_node(node)
571 elif target.is_root_node() and position in ("left", "right"):
572 self._make_sibling_of_root_node(node, target, position)
573 else:
574 if node.is_root_node():
575 self._move_root_node(node, target, position)
576 else:
577 self._move_child_node(node, target, position)
579 def move_node(self, node, target, position="last-child"):
580 """
581 Moves ``node`` relative to a given ``target`` node as specified
582 by ``position`` (when appropriate), by examining both nodes and
583 calling the appropriate method to perform the move.
585 A ``target`` of ``None`` indicates that ``node`` should be
586 turned into a root node.
588 Valid values for ``position`` are ``'first-child'``,
589 ``'last-child'``, ``'left'`` or ``'right'``.
591 ``node`` will be modified to reflect its new tree state in the
592 database.
594 This method explicitly checks for ``node`` being made a sibling
595 of a root node, as this is a special case due to our use of tree
596 ids to order root nodes.
598 NOTE: This is a low-level method; it does NOT respect
599 ``MPTTMeta.order_insertion_by``. In most cases you should just
600 move the node yourself by setting node.parent.
601 """
602 self._move_node(node, target, position=position)
603 node.save()
604 node_moved.send(
605 sender=node.__class__, instance=node, target=target, position=position
606 )
608 @delegate_manager
609 def root_node(self, tree_id):
610 """
611 Returns the root node of the tree with the given id.
612 """
613 return self._mptt_filter(tree_id=tree_id, parent=None).get()
615 @delegate_manager
616 def root_nodes(self):
617 """
618 Creates a ``QuerySet`` containing root nodes.
619 """
620 return self._mptt_filter(parent=None)
622 @delegate_manager
623 def rebuild(self):
624 """
625 Rebuilds all trees in the database table using `parent` link.
626 """
627 opts = self.model._mptt_meta
629 qs = self._mptt_filter(parent=None)
630 if opts.order_insertion_by:
631 qs = qs.order_by(*opts.order_insertion_by)
632 pks = qs.values_list("pk", flat=True)
634 rebuild_helper = self._rebuild_helper
635 idx = 0
636 for pk in pks:
637 idx += 1
638 rebuild_helper(pk, 1, idx)
640 rebuild.alters_data = True
642 @delegate_manager
643 def partial_rebuild(self, tree_id):
644 """
645 Partially rebuilds a tree i.e. It rebuilds only the tree with given
646 ``tree_id`` in database table using ``parent`` link.
647 """
648 opts = self.model._mptt_meta
650 qs = self._mptt_filter(parent=None, tree_id=tree_id)
651 if opts.order_insertion_by:
652 qs = qs.order_by(*opts.order_insertion_by)
653 pks = qs.values_list("pk", flat=True)
654 if not pks:
655 return
656 if len(pks) > 1:
657 raise RuntimeError(
658 "More than one root node with tree_id %d. That's invalid,"
659 " do a full rebuild." % tree_id
660 )
662 self._rebuild_helper(pks[0], 1, tree_id)
664 @delegate_manager
665 def build_tree_nodes(self, data, target=None, position="last-child"):
666 """
667 Load a tree from a nested dictionary for bulk insert, returning an
668 array of records. Use to efficiently insert many nodes within a tree
669 without an expensive `rebuild`.
671 ::
673 records = MyModel.objects.build_tree_nodes({
674 'id': 7,
675 'name': 'parent',
676 'children': [
677 {
678 'id': 8,
679 'parent_id': 7,
680 'name': 'child',
681 'children': [
682 {
683 'id': 9,
684 'parent_id': 8,
685 'name': 'grandchild',
686 }
687 ]
688 }
689 ]
690 })
691 MyModel.objects.bulk_create(records)
693 """
694 opts = self.model._mptt_meta
695 if target:
696 tree_id = target.tree_id
697 if position in ("left", "right"):
698 level = getattr(target, opts.level_attr)
699 if position == "left":
700 cursor = getattr(target, opts.left_attr)
701 else:
702 cursor = getattr(target, opts.right_attr) + 1
703 else:
704 level = getattr(target, opts.level_attr) + 1
705 if position == "first-child":
706 cursor = getattr(target, opts.left_attr) + 1
707 else:
708 cursor = getattr(target, opts.right_attr)
709 else:
710 tree_id = self._get_next_tree_id()
711 cursor = 1
712 level = 0
714 stack = []
716 def treeify(data, cursor=1, level=0):
717 data = dict(data)
718 children = data.pop("children", [])
719 node = self.model(**data)
720 stack.append(node)
721 setattr(node, opts.tree_id_attr, tree_id)
722 setattr(node, opts.level_attr, level)
723 setattr(node, opts.left_attr, cursor)
724 for child in children:
725 cursor = treeify(child, cursor=cursor + 1, level=level + 1)
726 cursor += 1
727 setattr(node, opts.right_attr, cursor)
728 return cursor
730 treeify(data, cursor=cursor, level=level)
732 if target:
733 self._create_space(2 * len(stack), cursor - 1, tree_id)
735 return stack
737 def _rebuild_helper(self, pk, left, tree_id, level=0):
738 opts = self.model._mptt_meta
739 right = left + 1
741 qs = self._mptt_filter(parent__pk=pk)
742 if opts.order_insertion_by:
743 qs = qs.order_by(*opts.order_insertion_by)
744 child_ids = qs.values_list("pk", flat=True)
746 rebuild_helper = self._rebuild_helper
747 for child_id in child_ids:
748 right = rebuild_helper(child_id, right, tree_id, level + 1)
750 qs = self.model._default_manager.db_manager(self.db).filter(pk=pk)
751 self._mptt_update(qs, left=left, right=right, level=level, tree_id=tree_id)
753 return right + 1
755 def _post_insert_update_cached_parent_right(self, instance, right_shift, seen=None):
756 setattr(
757 instance, self.right_attr, getattr(instance, self.right_attr) + right_shift
758 )
759 parent = cached_field_value(instance, self.parent_attr)
760 if parent:
761 if not seen:
762 seen = set()
763 seen.add(instance)
764 if parent in seen: 764 ↛ 766line 764 didn't jump to line 766, because the condition on line 764 was never true
765 # detect infinite recursion and throw an error
766 raise InvalidMove
767 self._post_insert_update_cached_parent_right(parent, right_shift, seen=seen)
769 def _calculate_inter_tree_move_values(self, node, target, position):
770 """
771 Calculates values required when moving ``node`` relative to
772 ``target`` as specified by ``position``.
773 """
774 left = getattr(node, self.left_attr)
775 level = getattr(node, self.level_attr)
776 target_left = getattr(target, self.left_attr)
777 target_right = getattr(target, self.right_attr)
778 target_level = getattr(target, self.level_attr)
780 if position == "last-child" or position == "first-child": 780 ↛ 787line 780 didn't jump to line 787, because the condition on line 780 was never false
781 if position == "last-child": 781 ↛ 784line 781 didn't jump to line 784, because the condition on line 781 was never false
782 space_target = target_right - 1
783 else:
784 space_target = target_left
785 level_change = level - target_level - 1
786 parent = target
787 elif position == "left" or position == "right":
788 if position == "left":
789 space_target = target_left - 1
790 else:
791 space_target = target_right
792 level_change = level - target_level
793 parent = getattr(target, self.parent_attr)
794 else:
795 raise ValueError(_("An invalid position was given: %s.") % position)
797 left_right_change = left - space_target - 1
799 right_shift = 0
800 if parent: 800 ↛ 803line 800 didn't jump to line 803, because the condition on line 800 was never false
801 right_shift = 2 * (node.get_descendant_count() + 1)
803 return space_target, level_change, left_right_change, parent, right_shift
805 def _close_gap(self, size, target, tree_id):
806 """
807 Closes a gap of a certain ``size`` after the given ``target``
808 point in the tree identified by ``tree_id``.
809 """
810 self._manage_space(-size, target, tree_id)
812 def _create_space(self, size, target, tree_id):
813 """
814 Creates a space of a certain ``size`` after the given ``target``
815 point in the tree identified by ``tree_id``.
816 """
817 self._manage_space(size, target, tree_id)
819 def _create_tree_space(self, target_tree_id, num_trees=1):
820 """
821 Creates space for a new tree by incrementing all tree ids
822 greater than ``target_tree_id``.
823 """
824 qs = self._mptt_filter(tree_id__gt=target_tree_id)
825 self._mptt_update(qs, tree_id=F(self.tree_id_attr) + num_trees)
826 self.tree_model._mptt_track_tree_insertions(target_tree_id + 1, num_trees)
828 def _get_next_tree_id(self):
829 """
830 Determines the next largest unused tree id for the tree managed
831 by this manager.
832 """
833 max_tree_id = list(self.aggregate(Max(self.tree_id_attr)).values())[0]
834 max_tree_id = max_tree_id or 0
835 return max_tree_id + 1
837 def _inter_tree_move_and_close_gap(
838 self, node, level_change, left_right_change, new_tree_id
839 ):
840 """
841 Removes ``node`` from its current tree, with the given set of
842 changes being applied to ``node`` and its descendants, closing
843 the gap left by moving ``node`` as it does so.
844 """
845 connection = self._get_connection(instance=node)
846 qn = connection.ops.quote_name
848 opts = self.model._meta
849 inter_tree_move_query = """
850 UPDATE %(table)s
851 SET %(level)s = CASE
852 WHEN %(left)s >= %%s AND %(left)s <= %%s
853 THEN %(level)s - %%s
854 ELSE %(level)s END,
855 %(tree_id)s = CASE
856 WHEN %(left)s >= %%s AND %(left)s <= %%s
857 THEN %%s
858 ELSE %(tree_id)s END,
859 %(left)s = CASE
860 WHEN %(left)s >= %%s AND %(left)s <= %%s
861 THEN %(left)s - %%s
862 WHEN %(left)s > %%s
863 THEN %(left)s - %%s
864 ELSE %(left)s END,
865 %(right)s = CASE
866 WHEN %(right)s >= %%s AND %(right)s <= %%s
867 THEN %(right)s - %%s
868 WHEN %(right)s > %%s
869 THEN %(right)s - %%s
870 ELSE %(right)s END
871 WHERE %(tree_id)s = %%s""" % {
872 "table": qn(self.tree_model._meta.db_table),
873 "level": qn(opts.get_field(self.level_attr).column),
874 "left": qn(opts.get_field(self.left_attr).column),
875 "tree_id": qn(opts.get_field(self.tree_id_attr).column),
876 "right": qn(opts.get_field(self.right_attr).column),
877 }
879 left = getattr(node, self.left_attr)
880 right = getattr(node, self.right_attr)
881 gap_size = right - left + 1
882 gap_target_left = left - 1
883 params = [
884 left,
885 right,
886 level_change,
887 left,
888 right,
889 new_tree_id,
890 left,
891 right,
892 left_right_change,
893 gap_target_left,
894 gap_size,
895 left,
896 right,
897 left_right_change,
898 gap_target_left,
899 gap_size,
900 getattr(node, self.tree_id_attr),
901 ]
903 cursor = connection.cursor()
904 cursor.execute(inter_tree_move_query, params)
906 def _make_child_root_node(self, node, new_tree_id=None):
907 """
908 Removes ``node`` from its tree, making it the root node of a new
909 tree.
911 If ``new_tree_id`` is not specified a new tree id will be
912 generated.
914 ``node`` will be modified to reflect its new tree state in the
915 database.
916 """
917 left = getattr(node, self.left_attr)
918 right = getattr(node, self.right_attr)
919 level = getattr(node, self.level_attr)
920 if not new_tree_id:
921 new_tree_id = self._get_next_tree_id()
922 left_right_change = left - 1
924 self._inter_tree_move_and_close_gap(node, level, left_right_change, new_tree_id)
926 # Update the node to be consistent with the updated
927 # tree in the database.
928 setattr(node, self.left_attr, left - left_right_change)
929 setattr(node, self.right_attr, right - left_right_change)
930 setattr(node, self.level_attr, 0)
931 setattr(node, self.tree_id_attr, new_tree_id)
932 setattr(node, self.parent_attr, None)
933 node._mptt_cached_fields[self.parent_attr] = None
935 def _make_sibling_of_root_node(self, node, target, position):
936 """
937 Moves ``node``, making it a sibling of the given ``target`` root
938 node as specified by ``position``.
940 ``node`` will be modified to reflect its new tree state in the
941 database.
943 Since we use tree ids to reduce the number of rows affected by
944 tree mangement during insertion and deletion, root nodes are not
945 true siblings; thus, making an item a sibling of a root node is
946 a special case which involves shuffling tree ids around.
947 """
948 if node == target:
949 raise InvalidMove(_("A node may not be made a sibling of itself."))
951 opts = self.model._meta
952 tree_id = getattr(node, self.tree_id_attr)
953 target_tree_id = getattr(target, self.tree_id_attr)
955 if node.is_child_node():
956 if position == "left":
957 space_target = target_tree_id - 1
958 new_tree_id = target_tree_id
959 elif position == "right":
960 space_target = target_tree_id
961 new_tree_id = target_tree_id + 1
962 else:
963 raise ValueError(_("An invalid position was given: %s.") % position)
965 self._create_tree_space(space_target)
966 if tree_id > space_target:
967 # The node's tree id has been incremented in the
968 # database - this change must be reflected in the node
969 # object for the method call below to operate on the
970 # correct tree.
971 setattr(node, self.tree_id_attr, tree_id + 1)
972 self._make_child_root_node(node, new_tree_id)
973 else:
974 if position == "left":
975 if target_tree_id > tree_id:
976 left_sibling = target.get_previous_sibling()
977 if node == left_sibling:
978 return
979 new_tree_id = getattr(left_sibling, self.tree_id_attr)
980 lower_bound, upper_bound = tree_id, new_tree_id
981 shift = -1
982 else:
983 new_tree_id = target_tree_id
984 lower_bound, upper_bound = new_tree_id, tree_id
985 shift = 1
986 elif position == "right":
987 if target_tree_id > tree_id:
988 new_tree_id = target_tree_id
989 lower_bound, upper_bound = tree_id, target_tree_id
990 shift = -1
991 else:
992 right_sibling = target.get_next_sibling()
993 if node == right_sibling:
994 return
995 new_tree_id = getattr(right_sibling, self.tree_id_attr)
996 lower_bound, upper_bound = new_tree_id, tree_id
997 shift = 1
998 else:
999 raise ValueError(_("An invalid position was given: %s.") % position)
1001 connection = self._get_connection(instance=node)
1002 qn = connection.ops.quote_name
1004 root_sibling_query = """
1005 UPDATE %(table)s
1006 SET %(tree_id)s = CASE
1007 WHEN %(tree_id)s = %%s
1008 THEN %%s
1009 ELSE %(tree_id)s + %%s END
1010 WHERE %(tree_id)s >= %%s AND %(tree_id)s <= %%s""" % {
1011 "table": qn(self.tree_model._meta.db_table),
1012 "tree_id": qn(opts.get_field(self.tree_id_attr).column),
1013 }
1015 cursor = connection.cursor()
1016 cursor.execute(
1017 root_sibling_query,
1018 [tree_id, new_tree_id, shift, lower_bound, upper_bound],
1019 )
1020 setattr(node, self.tree_id_attr, new_tree_id)
1022 def _manage_space(self, size, target, tree_id):
1023 """
1024 Manages spaces in the tree identified by ``tree_id`` by changing
1025 the values of the left and right columns by ``size`` after the
1026 given ``target`` point.
1027 """
1028 if self.tree_model._mptt_is_tracking: 1028 ↛ 1029line 1028 didn't jump to line 1029, because the condition on line 1028 was never true
1029 self.tree_model._mptt_track_tree_modified(tree_id)
1030 else:
1031 connection = self._get_connection()
1032 qn = connection.ops.quote_name
1034 opts = self.model._meta
1035 space_query = """
1036 UPDATE %(table)s
1037 SET %(left)s = CASE
1038 WHEN %(left)s > %%s
1039 THEN %(left)s + %%s
1040 ELSE %(left)s END,
1041 %(right)s = CASE
1042 WHEN %(right)s > %%s
1043 THEN %(right)s + %%s
1044 ELSE %(right)s END
1045 WHERE %(tree_id)s = %%s
1046 AND (%(left)s > %%s OR %(right)s > %%s)""" % {
1047 "table": qn(self.tree_model._meta.db_table),
1048 "left": qn(opts.get_field(self.left_attr).column),
1049 "right": qn(opts.get_field(self.right_attr).column),
1050 "tree_id": qn(opts.get_field(self.tree_id_attr).column),
1051 }
1052 cursor = connection.cursor()
1053 cursor.execute(
1054 space_query, [target, size, target, size, tree_id, target, target]
1055 )
1057 def _move_child_node(self, node, target, position):
1058 """
1059 Calls the appropriate method to move child node ``node``
1060 relative to the given ``target`` node as specified by
1061 ``position``.
1062 """
1063 tree_id = getattr(node, self.tree_id_attr)
1064 target_tree_id = getattr(target, self.tree_id_attr)
1066 if tree_id == target_tree_id:
1067 self._move_child_within_tree(node, target, position)
1068 else:
1069 self._move_child_to_new_tree(node, target, position)
1071 def _move_child_to_new_tree(self, node, target, position):
1072 """
1073 Moves child node ``node`` to a different tree, inserting it
1074 relative to the given ``target`` node in the new tree as
1075 specified by ``position``.
1077 ``node`` will be modified to reflect its new tree state in the
1078 database.
1079 """
1080 left = getattr(node, self.left_attr)
1081 right = getattr(node, self.right_attr)
1082 level = getattr(node, self.level_attr)
1083 new_tree_id = getattr(target, self.tree_id_attr)
1085 (
1086 space_target,
1087 level_change,
1088 left_right_change,
1089 parent,
1090 new_parent_right,
1091 ) = self._calculate_inter_tree_move_values(node, target, position)
1093 tree_width = right - left + 1
1095 # Make space for the subtree which will be moved
1096 self._create_space(tree_width, space_target, new_tree_id)
1097 # Move the subtree
1098 self._inter_tree_move_and_close_gap(
1099 node, level_change, left_right_change, new_tree_id
1100 )
1102 # Update the node to be consistent with the updated
1103 # tree in the database.
1104 setattr(node, self.left_attr, left - left_right_change)
1105 setattr(node, self.right_attr, right - left_right_change)
1106 setattr(node, self.level_attr, level - level_change)
1107 setattr(node, self.tree_id_attr, new_tree_id)
1108 setattr(node, self.parent_attr, parent)
1110 node._mptt_cached_fields[self.parent_attr] = parent.pk
1112 def _move_child_within_tree(self, node, target, position):
1113 """
1114 Moves child node ``node`` within its current tree relative to
1115 the given ``target`` node as specified by ``position``.
1117 ``node`` will be modified to reflect its new tree state in the
1118 database.
1119 """
1120 left = getattr(node, self.left_attr)
1121 right = getattr(node, self.right_attr)
1122 level = getattr(node, self.level_attr)
1123 width = right - left + 1
1124 tree_id = getattr(node, self.tree_id_attr)
1125 target_left = getattr(target, self.left_attr)
1126 target_right = getattr(target, self.right_attr)
1127 target_level = getattr(target, self.level_attr)
1129 if position == "last-child" or position == "first-child":
1130 if node == target:
1131 raise InvalidMove(_("A node may not be made a child of itself."))
1132 elif left < target_left < right:
1133 raise InvalidMove(
1134 _("A node may not be made a child of any of its descendants.")
1135 )
1136 if position == "last-child":
1137 if target_right > right:
1138 new_left = target_right - width
1139 new_right = target_right - 1
1140 else:
1141 new_left = target_right
1142 new_right = target_right + width - 1
1143 else:
1144 if target_left > left:
1145 new_left = target_left - width + 1
1146 new_right = target_left
1147 else:
1148 new_left = target_left + 1
1149 new_right = target_left + width
1150 level_change = level - target_level - 1
1151 parent = target
1152 elif position == "left" or position == "right":
1153 if node == target:
1154 raise InvalidMove(_("A node may not be made a sibling of itself."))
1155 elif left < target_left < right:
1156 raise InvalidMove(
1157 _("A node may not be made a sibling of any of its descendants.")
1158 )
1159 if position == "left":
1160 if target_left > left:
1161 new_left = target_left - width
1162 new_right = target_left - 1
1163 else:
1164 new_left = target_left
1165 new_right = target_left + width - 1
1166 else:
1167 if target_right > right:
1168 new_left = target_right - width + 1
1169 new_right = target_right
1170 else:
1171 new_left = target_right + 1
1172 new_right = target_right + width
1173 level_change = level - target_level
1174 parent = getattr(target, self.parent_attr)
1175 else:
1176 raise ValueError(_("An invalid position was given: %s.") % position)
1178 left_boundary = min(left, new_left)
1179 right_boundary = max(right, new_right)
1180 left_right_change = new_left - left
1181 gap_size = width
1182 if left_right_change > 0:
1183 gap_size = -gap_size
1185 connection = self._get_connection(instance=node)
1186 qn = connection.ops.quote_name
1188 opts = self.model._meta
1189 # The level update must come before the left update to keep
1190 # MySQL happy - left seems to refer to the updated value
1191 # immediately after its update has been specified in the query
1192 # with MySQL, but not with SQLite or Postgres.
1193 move_subtree_query = """
1194 UPDATE %(table)s
1195 SET %(level)s = CASE
1196 WHEN %(left)s >= %%s AND %(left)s <= %%s
1197 THEN %(level)s - %%s
1198 ELSE %(level)s END,
1199 %(left)s = CASE
1200 WHEN %(left)s >= %%s AND %(left)s <= %%s
1201 THEN %(left)s + %%s
1202 WHEN %(left)s >= %%s AND %(left)s <= %%s
1203 THEN %(left)s + %%s
1204 ELSE %(left)s END,
1205 %(right)s = CASE
1206 WHEN %(right)s >= %%s AND %(right)s <= %%s
1207 THEN %(right)s + %%s
1208 WHEN %(right)s >= %%s AND %(right)s <= %%s
1209 THEN %(right)s + %%s
1210 ELSE %(right)s END
1211 WHERE %(tree_id)s = %%s""" % {
1212 "table": qn(self.tree_model._meta.db_table),
1213 "level": qn(opts.get_field(self.level_attr).column),
1214 "left": qn(opts.get_field(self.left_attr).column),
1215 "right": qn(opts.get_field(self.right_attr).column),
1216 "tree_id": qn(opts.get_field(self.tree_id_attr).column),
1217 }
1219 cursor = connection.cursor()
1220 cursor.execute(
1221 move_subtree_query,
1222 [
1223 left,
1224 right,
1225 level_change,
1226 left,
1227 right,
1228 left_right_change,
1229 left_boundary,
1230 right_boundary,
1231 gap_size,
1232 left,
1233 right,
1234 left_right_change,
1235 left_boundary,
1236 right_boundary,
1237 gap_size,
1238 tree_id,
1239 ],
1240 )
1242 # Update the node to be consistent with the updated
1243 # tree in the database.
1244 setattr(node, self.left_attr, new_left)
1245 setattr(node, self.right_attr, new_right)
1246 setattr(node, self.level_attr, level - level_change)
1247 setattr(node, self.parent_attr, parent)
1248 node._mptt_cached_fields[self.parent_attr] = parent.pk
1250 def _move_root_node(self, node, target, position):
1251 """
1252 Moves root node``node`` to a different tree, inserting it
1253 relative to the given ``target`` node as specified by
1254 ``position``.
1256 ``node`` will be modified to reflect its new tree state in the
1257 database.
1258 """
1259 left = getattr(node, self.left_attr)
1260 right = getattr(node, self.right_attr)
1261 level = getattr(node, self.level_attr)
1262 tree_id = getattr(node, self.tree_id_attr)
1263 new_tree_id = getattr(target, self.tree_id_attr)
1264 width = right - left + 1
1266 if node == target:
1267 raise InvalidMove(_("A node may not be made a child of itself."))
1268 elif tree_id == new_tree_id:
1269 raise InvalidMove(
1270 _("A node may not be made a child of any of its descendants.")
1271 )
1273 (
1274 space_target,
1275 level_change,
1276 left_right_change,
1277 parent,
1278 right_shift,
1279 ) = self._calculate_inter_tree_move_values(node, target, position)
1281 # Create space for the tree which will be inserted
1282 self._create_space(width, space_target, new_tree_id)
1284 # Move the root node, making it a child node
1285 connection = self._get_connection(instance=node)
1286 qn = connection.ops.quote_name
1288 opts = self.model._meta
1289 move_tree_query = """
1290 UPDATE %(table)s
1291 SET %(level)s = %(level)s - %%s,
1292 %(left)s = %(left)s - %%s,
1293 %(right)s = %(right)s - %%s,
1294 %(tree_id)s = %%s
1295 WHERE %(left)s >= %%s AND %(left)s <= %%s
1296 AND %(tree_id)s = %%s""" % {
1297 "table": qn(self.tree_model._meta.db_table),
1298 "level": qn(opts.get_field(self.level_attr).column),
1299 "left": qn(opts.get_field(self.left_attr).column),
1300 "right": qn(opts.get_field(self.right_attr).column),
1301 "tree_id": qn(opts.get_field(self.tree_id_attr).column),
1302 }
1304 cursor = connection.cursor()
1305 cursor.execute(
1306 move_tree_query,
1307 [
1308 level_change,
1309 left_right_change,
1310 left_right_change,
1311 new_tree_id,
1312 left,
1313 right,
1314 tree_id,
1315 ],
1316 )
1318 # Update the former root node to be consistent with the updated
1319 # tree in the database.
1320 setattr(node, self.left_attr, left - left_right_change)
1321 setattr(node, self.right_attr, right - left_right_change)
1322 setattr(node, self.level_attr, level - level_change)
1323 setattr(node, self.tree_id_attr, new_tree_id)
1324 setattr(node, self.parent_attr, parent)
1325 node._mptt_cached_fields[self.parent_attr] = parent.pk