如何用设计模式重构Python中多任务串行的冗长函数？

你的需求是把冗长的顺序执行函数拆成独立任务类并串联执行，责任链模式确实不适用——它的核心是「请求传递/可选处理」，而你需要的是「固定顺序执行全部任务」，最合适的是基于命令模式的流水线实现，每个任务封装成独立类，由一个调度类按顺序触发执行。

实现步骤 #

1. 定义任务抽象基类 #

先给所有任务统一接口，确保每个任务都有execute方法，还可以通过context参数在任务间传递共享数据（比如仓库路径、配置信息）：

from abc import ABC, abstractmethod

class BaseTask(ABC):
    @abstractmethod
    def execute(self, context=None):
        """执行任务，context用于传递任务间的共享数据"""
        pass

2. 实现具体任务类 #

每个子任务写成独立类，专注单一职责，比如你提到的克隆仓库、编辑文件、创建分支：

import subprocess

# 克隆仓库任务
class CloneRepoTask(BaseTask):
    def __init__(self, repo_url, target_dir):
        self.repo_url = repo_url
        self.target_dir = target_dir
    
    def execute(self, context=None):
        # 实际执行克隆命令
        subprocess.run(["git", "clone", self.repo_url, self.target_dir], check=True)
        # 将仓库路径存入context，供后续任务使用
        if context:
            context["repo_dir"] = self.target_dir
        return context

# 编辑文件任务
class EditFileTask(BaseTask):
    def __init__(self, file_path, content):
        self.file_path = file_path
        self.content = content
    
    def execute(self, context=None):
        if not context or "repo_dir" not in context:
            raise ValueError("必须先完成仓库克隆，获取仓库路径")
        full_file_path = f"{context['repo_dir']}/{self.file_path}"
        # 实际写入文件
        with open(full_file_path, "w", encoding="utf-8") as f:
            f.write(self.content)
        return context

# 创建新分支任务
class CreateBranchTask(BaseTask):
    def __init__(self, branch_name):
        self.branch_name = branch_name
    
    def execute(self, context=None):
        if not context or "repo_dir" not in context:
            raise ValueError("必须先完成仓库克隆，获取仓库路径")
        # 实际创建分支
        subprocess.run(
            ["git", "checkout", "-b", self.branch_name],
            cwd=context["repo_dir"],
            check=True
        )
        return context

3. 实现调度类（MainClass） #

这个类负责管理任务列表，按顺序执行所有任务，还可以处理异常、维护执行上下文：

class TaskPipeline:
    def __init__(self):
        self.tasks = []
    
    def add_task(self, task):
        if not isinstance(task, BaseTask):
            raise TypeError("添加的任务必须是BaseTask的子类")
        self.tasks.append(task)
    
    def run(self, initial_context=None):
        context = initial_context or {}
        for idx, task in enumerate(self.tasks):
            try:
                context = task.execute(context)
            except Exception as e:
                # 可根据需求选择终止执行或跳过，这里直接抛出异常终止
                raise RuntimeError(f"执行第{idx+1}个任务时出错: {str(e)}") from e
        return context

4. 使用示例 #

把任务按顺序加入流水线，一键执行：

if __name__ == "__main__":
    # 初始化各个任务
    clone_task = CloneRepoTask("https://github.com/example/repo.git", "./my-repo")
    edit_readme = EditFileTask("README.md", "# 重构后的项目\n这是用流水线模式拆分的任务")
    create_feature_branch = CreateBranchTask("feature/new-module")

    # 创建流水线并添加任务（顺序就是执行顺序）
    pipeline = TaskPipeline()
    pipeline.add_task(clone_task)
    pipeline.add_task(edit_readme)
    pipeline.add_task(create_feature_branch)

    # 执行所有任务
    pipeline.run()

方案优势 #

• 解耦易维护：每个任务只负责自己的逻辑，修改克隆逻辑不用动文件编辑的代码
• 灵活调整：修改执行顺序只需要调整add_task的调用顺序，新增任务只需要继承BaseTask实现execute
• 上下文传递：通过context安全传递任务间的共享数据，避免全局变量或硬编码
• 易于扩展：可以轻松添加日志、异常重试、任务撤销等功能，比如在TaskPipeline的run方法里加日志，或者给任务类加rollback方法

为什么不用责任链模式 #

责任链的核心是「请求在处理者之间传递，直到被处理」，适合需要动态决定是否处理请求的场景（比如权限校验、请求过滤），而你需要的是所有任务必须按固定顺序执行，显然流水线模式更匹配你的需求。

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