处理具有多个任务的python中的错误

Question

我的 python 工作流程中有多个任务，想知道处理错误的最佳方法是什么。

class Task1():
    is_ready = False

    def run(self):
        try:
            a = 0/0
            # some more operations
            self._is_ready = True
        except:
            print 'logging errors'


class Task2():
    _is_ready = False

    def run(self):
        try:
            a = 1
            # some more operations
            self._is_ready = True
        except:
            print 'logging errors'

class Workflow():
    def run(self, ):
        self.task1 = Task1()
        self.task2 = Task2()
        self.task1.run()
        if self.task1.is_ready:
            self.task2.run()

w = Workflow()
w.run()         

我基本上想根据每个任务的错误顺序运行每个任务。 IE;如果 task1 运行良好，则处理 task2..

您能否告诉我上述方法是否正确？

我总共有 10 个任务，并且认为添加多个 if 循环听起来不是一个好方法..

Answer 1

这里真的有两个问题。一个是关于如何按顺序排列任务，另一个是关于如果一个任务失败了如何打破顺序。

如果您想要任何类型的可扩展性，您将需要一个可迭代的任务，以便您可以在其上运行 for 循环。正如您自己所注意到的，使用嵌套的ifs 是完全不切实际的。基本结构在概念上是这样的：

tasks = [Task1(), Task2(), ...]
for task in tasks:
    task.run()
    if task.failed():
        break

循环的任何部分都不需要以书面形式出现。循环本身可以替换为any、all 或next。状态检查可以是属性检查、方法调用甚至是隐含的异常。

对于如何确定任务是否失败，您有多种选择：

1. 按照您当前的做法使用内部标志。确保标志在所有任务类中具有一致的名称（注意Task2 中的拼写错误_is_ready）。这有点矫枉过正，除非你有一个真正需要它的用例，因为它提供了冗余信息，而且不是很优雅。

2. 使用run 中的返回值。这更好，因为你可以写

   for task in tasks:
       if not task.run():
           break
   

   或者（正如@MichaelButscher 巧妙地建议的那样）

   all(task.run() for task in tasks)
   

   在任何一种情况下，您的任务都应如下所示：

   class Task1:
       def run(self):
           try:
               # Some stuff
           except SomeException:
               # Log error
               return False
           return True
   

3. 让错误从任务实现中传播：

   class Task1:
       def run(self):
           try:
               # Some stuff
           except SomeException:
               # Log error
               raise
   

   我比其他所有方法都更喜欢这种方法，因为这首先是例外情况。在这种情况下，您的循环将更加简约：

   for task in tasks:
       task.run()
   

   或者，但更隐蔽

   any(task.run() for task in tasks)
   

   甚至

   from collections import deque
   deque(task.run() for task in tasks, maxlen=0)
   

   后两个选项实际上仅供参考。如果你有例外，只需编写基本的for 循环：它足够优雅，而且迄今为止最不神秘。

最后，我会推荐另一个根本性的改变。如果您的任务本质上确实是任意的，那么您应该考虑允许任何不带参数的可调用对象作为任务。没有特别需要将自己限制在具有run 方法的类中。如果你需要一个任务类，你可以把你调用的方法run重命名为__call__，你所有的实例都可以用()操作符调用。代码在概念上看起来像这样：

class CallableClass:
    def __call__(self):
        try:
            # Do something
        except:
            # Log error
            raise

def callable_function():
    try:
        # Do something
    except:
        # Log error
        raise

for task in tasks:
    task()

Answer 2

如果run() 方法可以返回一个布尔成功值并且每个任务只有在之前成功的情况下才应该运行，那么它可以这样完成：

class Workflow():
    def run(self, ):
        task_list = (Task1(), Task2(), Task3(), ...)
        success = all(t.run()  for t in task_list)