在 Django 中创建项目管理应用程序 - DAG 和依赖项

Question

我正在使用项目管理应用程序。应用程序需要的一点是创建具有依赖关系的任务的可能性，即任务 2 依赖于任务 1，任务 3 依赖于任务 2 和任务 B，等等。这将生成一个 DAG（有向无环图）。

问题是如何使用 Django 默认的 ORM 来存储这个“图”。

我提供了一个使用多对多关系（Task with self）和触发器来避免创建循环的解决方案，但我仍然不知道如何实现一些操作，例如从单个节点。

有没有人知道如何很好地实现它？例如，这将允许执行关键路径方法的实现。

编辑：

使用@Daniel 提出的解决方案，我最终得到了以下功能。我们可以看到，为了获得整个图，对数据库进行了多次调用。此函数在循环期间也会进入无限循环。

def get_task_graph(task, visited_nodes = [], graph = {}):
    # Visited nodes are nodes that all parents were visited
    # print(task)

    
    
    previous_tasks = task.previous_tasks.all()

    
    still_have_unmarked_parents = set(previous_tasks).difference(set(visited_nodes))
    if still_have_unmarked_parents:
        print(f"Task {task} still have unmarked parents")
        
        for previous_task in still_have_unmarked_parents:
            

            visited_nodes, graph =  get_task_graph(previous_task, visited_nodes=visited_nodes, graph=graph)
    
    next_tasks = task.next_tasks.all()
 
    visited_nodes.append(task)
    print(f"Marking {task} as visited")
    graph[task] = []
    for next_task in next_tasks:
        if next_task not in graph[task]:
            graph[task].append(next_task)
    
        
        visited_nodes, graph = get_task_graph(next_task, visited_nodes=visited_nodes, graph=graph)
        print(f"\t Adding {next_task} to {task}")

    return visited_nodes, graph

谢谢

Answer 1

原答案：多对多关系

让我们创建一个简单的Task 模型：

class Task(models.Model):
   name = models.CharField(...)
   previous_tasks = models.ManyToMany('Task', ..., related_name='next_tasks')

现在让我们创建taks_a 和task_b - task_b 将依赖于task_a：

# create task_a, task_b:
task_a = Task.objects.create(name='First Task')
task_b = Task.objects.create(name='Second task')

# add task_a as a dependency for task_b:
task_b.previous_tasks.add([task_a])
task_b.save()

现在我们可以像这样访问这些任务：

# get the tasks to do after task_a is complete:
task_a = Task.objects.get(id='<task-a-id>')
tasks_to_do_after_task_a = task_a.next_tasks.all() # returns a queryset with <task_b>

# get the tasks we need to complete before task_b:
task_b = Task.objects.get(id='<task-b-id>')
tasks_that_task_b_depends_on = task_b.previous_tasks.all() # returns a queryset with <task a>

最后，让我们假设 task_a 实际上有几个不同的任务依赖于它：

# iterate over all the tasks that depend on task a:
for tasks in task_a.next_tasks.all():

     # do some logic, i.e. find longest task:
     ...

# filter for a specific next_task on task_a:
some_task = task_a.next_tasks.filter(some_keyword='some_value')

# assume we have a duration time-field on tasks:      
next_longest_task = task_a.next_tasks.order_by('-duration')[0] # gets the next task with the longest duration

您可以从那里根据需要实现寻路逻辑/其他算法。

编辑：递归访问字段

好的 - 使用上面的结构，我们可以像这样定义一个函数：

def get_task_list(task):

    # get the next_tasks queryset, it will either contain tasks or be an empty queryset:
    next_tasks = task.next_tasks.all()

    # check if the current task has any children:
    if next_tasks:

        # do some logic, check for longest tasks / add next tasks to a list
        ...

        # pass each next task back into the function:
        for next_task in next_tasks:
            return get_task_list(next_task)

    # exit if there are no more next tasks:
    else:

        # do some logic on exiting, calculate the time to complete all tasks:
        ...

        return <optional-content>

您可以对递归函数进行更多研究，但这应该可以帮助您入门。您可以在函数中向上查看以前的任务或向下查看下一个任务时变得聪明。您还可以将此作为方法添加到您的 Task 模型中，以便在任务实例上调用它。