Python 状态机（transitions）实践

pip3 install transitions

from transitions import Machine


class StatusEnum:
    READY = '待开始'
    PROCESSING = '进行中'
    FINISHED = '已完成'
    REREADY = '待再次开始'
    SKIPED = '已跳过'
    BACKED = '已驳回'


# 定义一个基础类
class LabelTask:
    states = [
        StatusEnum.READY, StatusEnum.PROCESSING, StatusEnum.FINISHED, StatusEnum.REREADY, StatusEnum.SKIPED,
        StatusEnum.BACKED
    ]
    # 状态流转转换关系，下面第一个对象意思是，触发动作为'开始'，允许操作的当前状态是StatusEnum.READY和StatusEnum.REREADY，
    # 变更后的状态是StatusEnum.PROCESSING，状态变更成功之后会调用 after_started 方法
    transitions = [{
        'trigger': '开始',
        'source': [StatusEnum.READY, StatusEnum.REREADY],
        'dest': StatusEnum.PROCESSING,
        'after': 'after_started'
    }, {
        'trigger': '完成',
        'source': StatusEnum.PROCESSING,
        'dest': StatusEnum.FINISHED,
        'conditions': ['can_completed'], # 执行状态变更的前置校验
        'after': 'after_completed'
    }, {
        'trigger': '打回',
        'source': StatusEnum.PROCESSING,
        'dest': StatusEnum.BACKED,
        'conditions': ['can_backed'],
        'after': 'after_backed'
    }, {
        'trigger': '跳过',
        'source': [StatusEnum.READY, StatusEnum.PROCESSING],
        'dest': StatusEnum.SKIPED,
        'conditions': ['can_skiped'],
        'after': 'after_skiped'
    }]
    
    def __init__(self, label_task_id, assignment_id, collection_params, api_request):
        pass

    def after_started(self):
        """
        执行'开始'动作之后的触发动作
        """
        pass

    def after_completed(self):
        """
        执行'完成'动作之后的触发动作
        """
        pass

    def after_backed(self):
        """
        执行打回'打回'之后的触发动作
        """
        pass

    def after_skiped(self):
        """
        执行打回'跳过'之后的触发动作
        """
        pass

    def can_completed(self):
        """
        允许操作'完成'动作的条件
        """
        if self.assignment_cat == '挑选':
            # 挑选任务没有上报信息, 直接把当前输入数据集文件传递给下一步
            self.dataset_path = self.api_request.get_dataset(self.assignment_dataset_id)['data']['attributes']['prefix']
            return True
        assert self.collection_params, '没有上报信息,不能完成'
        assert self.dataset_path, '上报信息必需包含新文件路径'
        return True

    def can_backed(self):
        """
        允许操作'打回'动作的条件
        """
        assert self.assignment_index > 0, '第一步不能打回'
        assert self.collection_params, '没有上报信息,不能打回'
        assert self.dataset_path, '上报信息必需包含新文件路径'
        return True

    def can_skiped(self):
        """
        允许操作'跳过'动作的条件
        """
        assert self.assignment_index == 0, '只有第一个阶段才能跳过'
        return True


label_task = LabelTask()
machine = Machine(
    label_task,
    states=label_task.states,
    transitions=label_task.transitions,
    initial=label_task.assignment_status, # 初始状态
    model_attribute='assignment_status', # 改为读取label_task.assignment_status这个值作为任务的状态
    auto_transitions=False # 禁止to_«state»()方法使用
)

# 触发任务改变任务额状态
label_task.trigger('开始')

print("当前任务的状态---", label_task.state)

{
        'trigger': '完成', # 表示触发动作接收‘完成’这个动作，即 label_task.trigger('完成')
        'source': StatusEnum.PROCESSING, # 当前任务的状态
        'dest': StatusEnum.FINISHED, # 变更之后的状态
        'conditions': ['can_completed'], # 执行状态变更的前置校验
        'after': 'after_completed' # 状态变更下一个状态之后的触发方法
    }