python - 如何在DDD中创建管道流程？

Question

我在使用 DDD 定义管道流程时遇到一些困难。我的场景如下:

有 3 个有界上下文:

• 患者数据管理器:负责检索和管理整个患者的数据，如 mrna、mirna、dna 甲基化等。

• 注释数据管理:即基因、mirna、蛋白质等注释。

• 分析器:这是一个特定的分析器，它使用一些患者的数据并提取一些有用的信息。此外，用户可以创建管道并创建 puglin，定义一些附加步骤来使用患者数据进行处理。每个步骤都是唯一的，并且只执行一项工作，如下所示:

  管道1:任务1 -> 任务2 -> 任务3 -> 任务4管道2:任务1->任务3->任务5->任务6

每个任务都会生成一些数据，因此任务必须管理数据，读取和写入数据。

我希望 reduce task 和任务与患者数据管理器上下文之间的耦合。

例如:由于数据量的原因，task1 必须获取患者数据管理器中的数据。 task1 获取一些数据，处理它，然后获取更多数据并再次处理。此过程导致任务1 保存一些数据。

然后，管道中的下一个任务是任务 3。 Task3从task1获取全部数据。 taks3 处理数据，产生另一个数据。任务3拯救了他们。管道已完成。

请注意，管道是在运行之前配置的，并且每个任务都需要一些数据才能工作并生成一些数据。

我考虑使用命令模式在患者数据管理器上下文中获取数据。关于管道，它有工厂来定义和配置管道，每个任务都有一个用于获取要处理的数据的命令和用于保存其数据的命令。

我知道命令的紧密耦合。我想过使用消息来使用编排模式，但是由于管道和患者数据管理器中的数据量以及每个任务结果的数据量，这确实很难实现。

这样做可以吗？还有其他更好的解决方案吗？

代码是用Python编写的。

非常感谢!

Answer 1

DDD 不会以任何方式与管道冲突。这个问题有点假设，但我会尽力回答。从技术上讲，您可以只拥有一个执行特定于域的任务的通用任务界面。例如，请参阅下面的样板。

class DomainTask1:
    def __init__(self):
        pass

    def run(self):
        # do something

class DomainTask2:
    def __init__(self):
        pass

    def run(self):
        # do something else

def run_pipeline(pipeline):
    for task in pipeline:
        task.run()

pipeline = [
    DomainTask1(),
    DomainTask2()
]

run_pipeline(pipeline)