Python中基于root_task_id构建Task对象层级的代码错误排查

问题 #

我有一个包含任务表的SQLite数据库，该表通过root_task_id字段实现任务嵌套：例如id为1的任务是id为19的任务的根任务，id为19的任务是id为20的任务的根任务，即id为20的任务是id为19的子任务，id为19的任务是id为1的子任务。

将数据从数据库读取到Python程序后，我希望通过以下定义的Task类对象构建任务层级结构：

class Task:
     def __init__(self, task_id, root_task_id, task_name, task_text, task_is_executed, subtasks = []):
        self.id = task_id
        self.root_task_id = root_task_id
        self.name = task_name
        self.text = task_text
        if task_is_executed == 0:
            self.is_executed = "No executed"
        else:
            self.is_executed = "EXECUTED"
        self.subtasks = subtasks

层级结构需通过Task对象的subtasks属性实现，即根据SQL表的root_task_id字段将任务作为子任务嵌套。我编写了如下代码，但存在无法定位的错误：

def wrap_up_task_recursive(task: Task, tasks: list[Task], depth = 0):
    for root_task in tasks:
        print(f"root_task with id: {root_task.id}, is root? for task with id: {task.id}, that has root task with id: {task.root_task_id}?")
        print(depth)
        if task.root_task_id == root_task.id:
            print("Yes")
            root_task.subtasks.append(task)
            break
        else:
            if len(root_task.subtasks) != 0:
                wrap_up_task_recursive(task=task, tasks=root_task.subtasks, depth=depth + 1)


def convertTasksToUserFrandlyFormat2(all_tasks: list[Task]) -> list[Task]:
    root_tasks: list[Task] = []
    root_tasks_indexes: list[int] = []
    for index, task in enumerate(all_tasks):
        if task.root_task_id == None:
            root_tasks.append(task)
            root_tasks_indexes.append(index)
    root_tasks_indexes.sort(reverse=True)
    print(f"Root tasks indexes: {root_tasks_indexes}")
    for index in root_tasks_indexes:
        all_tasks.pop(index)
    print(f"**********root_tasks**********")
    for rt in root_tasks:
        print(f"id: {rt.id}")
        print(f"root_id: {rt.root_task_id}")
        print(f"sub: {len(rt.subtasks)}")
    print(f"**********Subtasks**********")
    for t in all_tasks:
        print(f"id: {t.id}")
        print(f"root_id: {t.root_task_id}")
        print(f"sub: {len(rt.subtasks)}")
    
    for task in all_tasks:
        wrap_up_task_recursive(task=task, tasks=root_tasks)
    return root_tasks

def indentation(depth: int) -> str:
    resultStr = ""
    for _ in range(depth):
        resultStr += "  "
    return resultStr

def recursiveReading(task: Task, depth = 0):
    print(f"{indentation(depth)} id = {task.id}")
    print(task.subtasks)
    print(len(task.subtasks))

def readTasksToUserFrandlyFormat(all_tasks: list[Task]):
    for task in all_tasks:
        recursiveReading(task)


task1 = Task(task_id=1, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task2 = Task(task_id=14, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task3 = Task(task_id=15, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task4 = Task(task_id=16, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task5 = Task(task_id=17, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task6 = Task(task_id=18, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task7 = Task(task_id=19, root_task_id=1, task_name="New task", task_text="This is new task", task_is_executed=0)
task8 = Task(task_id=20, root_task_id=19, task_name="New task", task_text="This is new task", task_is_executed=0)
all_tasks = [task1, task2, task3, task4, task5, task6, task7, task8]
h_all_tasks = convertTasksToUserFrandlyFormat2(all_tasks)
readTasksToUserFrandlyFormat(h_all_tasks)

控制台输出显示，子任务被重复添加到多个根任务的subtasks列表中，每个根任务的subtasks都包含重复的任务实例，请求排查并修正该问题。

问题分析 #

1. Task类的可变默认参数陷阱：Python中如果将可变对象（比如[]）作为函数默认参数，会导致所有实例共享同一个对象。你的Task类__init__方法中subtasks = []就是这个问题，所有Task实例的默认subtasks指向同一个列表，添加子任务时会被所有实例共享。
2. 递归函数未终止后续遍历：wrap_up_task_recursive函数中，递归进入子层级找到父任务后，外层循环不会停止，会继续遍历其他根任务，导致同一个任务被多次添加到不同根任务的子列表中。
3. 打印代码变量误用：convertTasksToUserFrandlyFormat2函数中打印子任务的sub长度时，误用了外层循环的rt变量，应该用当前遍历的t变量。

修正后的代码 #

1. 修复Task类 #

class Task:
     def __init__(self, task_id, root_task_id, task_name, task_text, task_is_executed, subtasks=None):
        self.id = task_id
        self.root_task_id = root_task_id
        self.name = task_name
        self.text = task_text
        self.is_executed = "No executed" if task_is_executed == 0 else "EXECUTED"
        # 避免可变默认参数的陷阱
        self.subtasks = subtasks if subtasks is not None else []

2. 修复递归函数 #

修改递归函数，让它返回是否找到父任务的布尔值，找到后立即终止后续遍历：

def wrap_up_task_recursive(task: Task, tasks: list[Task]) -> bool:
    for root_task in tasks:
        if task.root_task_id == root_task.id:
            root_task.subtasks.append(task)
            return True  # 找到父任务，返回True终止查找
        # 递归查找当前任务的子层级
        if wrap_up_task_recursive(task, root_task.subtasks):
            return True
    return False  # 未找到

3. 修复转换函数中的打印错误 #

def convertTasksToUserFrandlyFormat2(all_tasks: list[Task]) -> list[Task]:
    root_tasks: list[Task] = []
    root_tasks_indexes: list[int] = []
    for index, task in enumerate(all_tasks):
        if task.root_task_id is None:
            root_tasks.append(task)
            root_tasks_indexes.append(index)
    root_tasks_indexes.sort(reverse=True)
    print(f"Root tasks indexes: {root_tasks_indexes}")
    for index in root_tasks_indexes:
        all_tasks.pop(index)
    
    print(f"**********root_tasks**********")
    for rt in root_tasks:
        print(f"id: {rt.id}, root_id: {rt.root_task_id}, sub_count: {len(rt.subtasks)}")
    
    print(f"**********Subtasks**********")
    for t in all_tasks:
        print(f"id: {t.id}, root_id: {t.root_task_id}, sub_count: {len(t.subtasks)}")
    
    for task in all_tasks:
        wrap_up_task_recursive(task, root_tasks)
    return root_tasks

4. 完善递归打印函数 #

让打印函数能递归展示所有子任务：

def recursiveReading(task: Task, depth = 0):
    indent = "  " * depth
    print(f"{indent}id = {task.id}, is_executed = {task.is_executed}")
    for subtask in task.subtasks:
        recursiveReading(subtask, depth + 1)

完整测试代码 #

class Task:
     def __init__(self, task_id, root_task_id, task_name, task_text, task_is_executed, subtasks=None):
        self.id = task_id
        self.root_task_id = root_task_id
        self.name = task_name
        self.text = task_text
        self.is_executed = "No executed" if task_is_executed == 0 else "EXECUTED"
        self.subtasks = subtasks if subtasks is not None else []

def wrap_up_task_recursive(task: Task, tasks: list[Task]) -> bool:
    for root_task in tasks:
        if task.root_task_id == root_task.id:
            root_task.subtasks.append(task)
            return True
        if wrap_up_task_recursive(task, root_task.subtasks):
            return True
    return False

def convertTasksToUserFrandlyFormat2(all_tasks: list[Task]) -> list[Task]:
    root_tasks: list[Task] = []
    root_tasks_indexes: list[int] = []
    for index, task in enumerate(all_tasks):
        if task.root_task_id is None:
            root_tasks.append(task)
            root_tasks_indexes.append(index)
    root_tasks_indexes.sort(reverse=True)
    print(f"Root tasks indexes: {root_tasks_indexes}")
    for index in root_tasks_indexes:
        all_tasks.pop(index)
    
    print(f"**********root_tasks**********")
    for rt in root_tasks:
        print(f"id: {rt.id}, root_id: {rt.root_task_id}, sub_count: {len(rt.subtasks)}")
    
    print(f"**********Subtasks**********")
    for t in all_tasks:
        print(f"id: {t.id}, root_id: {t.root_task_id}, sub_count: {len(t.subtasks)}")
    
    for task in all_tasks:
        wrap_up_task_recursive(task, root_tasks)
    return root_tasks

def recursiveReading(task: Task, depth = 0):
    indent = "  " * depth
    print(f"{indent}id = {task.id}, is_executed = {task.is_executed}")
    for subtask in task.subtasks:
        recursiveReading(subtask, depth + 1)

def readTasksToUserFrandlyFormat(all_tasks: list[Task]):
    for task in all_tasks:
        recursiveReading(task)

# 测试数据
task1 = Task(task_id=1, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task2 = Task(task_id=14, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task3 = Task(task_id=15, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task4 = Task(task_id=16, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task5 = Task(task_id=17, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task6 = Task(task_id=18, root_task_id=None, task_name="New task", task_text="This is new task", task_is_executed=0)
task7 = Task(task_id=19, root_task_id=1, task_name="New task", task_text="This is new task", task_is_executed=0)
task8 = Task(task_id=20, root_task_id=19, task_name="New task", task_text="This is new task", task_is_executed=0)
all_tasks = [task1, task2, task3, task4, task5, task6, task7, task8]

h_all_tasks = convertTasksToUserFrandlyFormat2(all_tasks)
print("\n**********层级结构展示**********")
readTasksToUserFrandlyFormat(h_all_tasks)

内容的提问来源于stack exchange，提问作者MaxFactorHub