如何使用django-treebeard的MP_Node预取子节点？

Question

我正在使用django-rest-framework和django-treebeard开发一个具有分层数据结构的应用程序。我的（简化后的）主模型如下：

class Task(MP_Node):
    name = models.CharField(_('name'), max_length=64)
    started = models.BooleanField(default=True)

我正在努力实现的目标是列出所有根节点的列表视图，其中显示额外字段（例如所有子项是否已启动）。为此，我指定了一个视图：

class TaskViewSet(viewsets.ViewSet):

    def retrieve(self, request, pk=None):
        queryset = Task.get_tree().filter(depth=1, job__isnull=True)
        operation = get_object_or_404(queryset, pk=pk)
        serializer = TaskSerializer(operation)
        return Response(serializer.data)

和序列化器

class TaskSerializer(serializers.ModelSerializer):
    are_children_started = serializers.SerializerMethodField()

    def get_are_children_started(self, obj):
        return all(task.started for task in Task.get_tree(obj))

这一切都能正常工作，我也得到了预期的结果。然而，我遇到了一个N+1查询问题，对于每个根任务，我需要单独获取所有子项。通常可以使用prefetch_related来解决此问题，但由于我使用了django-treebeard中的Materialized Path结构，因此任务模型之间没有Django关系，所以prefetch_related无法直接处理。我尝试使用自定义Prefetch对象，但由于这仍然需要Django关系路径，因此我无法使其正常工作。

我的当前想法是通过添加指向其根节点的外键来扩展Task模型，如下所示：

root_node = models.ForeignKey('self', null=True,
                              related_name='descendant_tasks',
                              verbose_name=_('root task')
                              )

为了使MP关系明确，以便进行查询。然而，这种方法似乎有点不够DRY，因此我想知道是否有其他建议来解决它。

- Matthijs Brouns

Answer 1

在最后，我确实添加了一个外键到每个任务，指向其根节点，就像这样：

root_node = models.ForeignKey('self', null=True,
                          related_name='descendant_tasks',
                          verbose_name=_('root task')
                          )

我更新了我的任务模型的保存方法，以确保我始终指向正确的根节点。

def save(self, force_insert=False, force_update=False, using=None, update_fields=None):
    try:
        self.root_task = self.get_root()
    except ObjectDoesNotExist:
        self.root_task = None

    return super(Task, self).save(force_insert=False, force_update=False, using=None,
                                  update_fields=None
                                  )

这让我能够使用prefetch_related('descendants')来简单地预获取所有后代。

每当我需要以嵌套方式获取后代时，我使用以下函数将扁平化的后代列表再次嵌套。

def build_nested(tasks):

    def get_basepath(path, depth):
        return path[0:depth * Task.steplen]

    container, link = [], {}
    for task in sorted(tasks, key=attrgetter('depth')):
        depth = int(len(task.path) / Task.steplen)
        try:
            parent_path = get_basepath(task.path, depth - 1)
            parent_obj = link[parent_path]
            if not hasattr(parent_obj, 'sub_tasks'):
                parent_obj.sub_tasks = []
            parent_obj.sub_tasks.append(task)
        except KeyError:  # Append it as root task if no parent exists
            container.append(task)

        link[task.path] = task

    return container

- Matthijs Brouns

Answer 2

如果您想避免使用外键，可以在查询集上迭代，并在内存中重新创建树形结构。

在我的情况下，我想要一个模板标签（类似于 django-mptt 的 recursetree 模板标签），以仅使用一个数据库查询显示多个级别的嵌套页面。基本上是复制 mptt.utils.get_cached_trees，最终得到了这个：

def get_cached_trees(queryset: QuerySet) -> list:
    """Return top-most pages / roots.

    Each page will have its children stored in `_cached_children` attribute
    and its parent in `_cached_parent`. This avoids having to query the database.
    """
    top_nodes: list = []
    path: list = []
    for obj in queryset:
        obj._cached_children = []
        if obj.depth == queryset[0].depth:
            add_top_node(obj, top_nodes, path)
        else:
            while not is_child_of(obj, parent := path[-1]):
                path.pop()
            add_child(parent, obj)

        if obj.numchild:
            path.append(obj)

    return top_nodes

def add_top_node(obj: MP_Node, top_nodes: list, path: list) -> None:
    top_nodes.append(obj)
    path.clear()

def add_child(parent: MP_Node, obj: MP_Node) -> None:
    obj._cached_parent = parent
    parent._cached_children.append(obj)

def is_child_of(child: MP_Node, parent: MP_Node) -> bool:
    """Return whether `child` is a sub page of `parent` without database query.

    `_get_children_path_interval` is an internal method of MP_Node.
    """
    start, end = parent._get_children_path_interval(parent.path)
    return start < child.path < end

可以这样使用，以避免可怕的N+1查询问题：

for page in get_cached_trees(queryset):
    for child in page._cached_children:
        ... 

- jnns