python - 用于记录的重复包装函数

Question

我正在编写一个需要使用外部库中的类的脚本，做一些对该类的实例进行操作，然后对更多实例重复。

像这样:

import some_library

work_queue = get_items()

for item in work_queue:
  some_object = some_library.SomeClass(item)
  operation_1(some_object)
  # ...
  operation_N(some_object)

但是，循环体中的每个操作都可能引发一些不同的异常。当这些发生时，我需要记录它们并跳到下一个项目。如果他们提出一些意外的异常我需要在崩溃之前记录下来。

我可以捕获主循环中的所有异常，但这会掩盖它的作用。所以我发现自己写了一堆看起来有点相似的包装函数:

def wrapper_op1(some_object):
  try:
    some_object.method_1()
  except (some_library.SomeOtherError, ValueError) as error_message:
    logger.error("Op1 error on {}".format(some_object.friendly_name))
    return False
  except Exception as error_message:
    logger.error("Unknown error during op1 on {} - crashing: {}".format(some_object.friendly_name, error_message))
    raise
  else:
    return True

# Notice there is a different tuple of anticipated exceptions
# and the message formatting is different
def wrapper_opN(some_object):
  try:
    some_function(some_object.some_attr)
  except (RuntimeError, AttributeError) as error_message:
    logger.error("OpN error on {} with {}".format(some_object.friendly_name, some_object.some_attr, error_message))
    return False
  except Exception as error_message:
    logger.error("Unknown error during opN on {} with {} - crashing: {}".(some_object.friendly_name, some_object.some_attr, error_message))
    raise
  else:
    return True

并将我的主循环修改为:

for item in work_queue:
  some_object = some_library.SomeClass(item)
  if not wrapper_op1(some_object):
    continue
  # ...
  if not wrapper_opN(some_object):
    continue

这完成了工作，但感觉就像是大量的复制和粘贴编程 wrapper 。最好是编写一个装饰器函数，它可以做所有尝试...除了...其他事情，这样我就可以做:

@ logged_call(known_exception, known_error_message, unknown_error_message)
def wrapper_op1(some_object):
  some_object.method_1()

如果操作成功，包装器将返回 True，捕获已知异常并以指定的格式记录，并在重新引发之前捕获任何未知的记录异常。

但是，我似乎无法理解如何使错误消息起作用——我可以使用固定字符串来做到这一点:

def logged_call(known_exceptions, s_err, s_fatal):
  def decorate(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
      try:
        f(*args, **kwargs)
      # How to get parameters from args in the message?
      except known_exceptions as error:
        print(s_err.format(error))
        return False
      except Exception as error:
        print(s_fatal.format(error))
        raise
      else:
        return True
    return wrapper
  return decorate

但是，我的错误消息需要获取属于修饰函数的属性。

是否有一些 Pythonic 的方法可以让它工作？或者使用不同的模式在处理可能以已知方式失败的函数时？

Answer 1

下面的 tryblock 函数可用于此目的，实现您的 logged_call 概念并减少代码总量，假设您有足够的检查来克服装饰器执行。不习惯使用 Python 进行函数式编程的人可能会发现这比简单地写出 try block 更难理解。简单，就像很多事情一样，在旁观者的眼中。

Python 2.7 不使用导入。这使用 exec 语句创建一个自定义的 try block ，该 block 以功能模式工作。

def tryblock(tryf, *catchclauses, **otherclauses):
  u'return a general try-catch-else-finally block as a function'
  elsef = otherclauses.get('elsef', None)
  finallyf = otherclauses.get('finallyf', None)
  namespace = {'tryf': tryf, 'elsef': elsef, 'finallyf': finallyf, 'func': []}
  for pair in enumerate(catchclauses):
    namespace['e%s' % (pair[0],)] = pair[1][0]
    namespace['f%s' % (pair[0],)] = pair[1][1]
  source = []
  add = lambda indent, line: source.append(' ' * indent + line)
  add(0, 'def block(*args, **kwargs):')
  add(1,   "u'generated function executing a try block'")
  add(1,   'try:')
  add(2,     '%stryf(*args, **kwargs)' % ('return ' if otherclauses.get('returnbody', elsef is None) else '',))
  for index in xrange(len(catchclauses)):
    add(1, 'except e%s as ex:' % (index,))
    add(2,   'return f%s(ex, *args, **kwargs)' % (index,))
  if elsef is not None:
    add(1,   'else:')
    add(2,     'return elsef(*args, **kwargs)')
  if finallyf is not None:
    add(1,   'finally:')
    add(2,     '%sfinallyf(*args, **kwargs)' % ('return ' if otherclauses.get('returnfinally', False) else '',))
  add(0, 'func.append(block)')
  exec '\n'.join(source) in namespace
  return namespace['func'][0]

这个 tryblock 函数足够通用，可以放入一个公共(public)库中，因为它并不特定于您的检查逻辑。向其中添加您的 logged_call 装饰器，实现为(此处导入一个):

import functools
resultof = lambda func: func()  # @ token must be followed by an identifier
@resultof
def logged_call():
  truism = lambda *args, **kwargs: True
  def raisef(ex, *args, **kwargs):
    raise ex
  def impl(exlist, err, fatal):
    return lambda func: \
      functools.wraps(func)(tryblock(func,
          (exlist, lambda ex, *args, **kwargs: err(ex, *args, **kwargs) and False),
          (Exception, lambda ex, *args, **kwargs: fatal(ex, *args, **kwargs) and raisef(ex))),
          elsef=truism)
  return impl  # impl therefore becomes logged_call

使用 logged_call 实现后，您的两个健全性检查如下所示:

op1check = logged_call((some_library.SomeOtherError, ValueError),
    lambda _, obj: logger.error("Op1 error on {}".format(obj.friendly_name)),
    lambda ex, obj: logger.error("Unknown error during op1 on {} - crashing: {}".format(obj.friendly_name, ex.message)))

opNcheck = logged_call((RuntimeError, AttributeError),
    lambda ex, obj: logger.error("OpN error on {} with {}".format(obj.friendly_name, obj.some_attr, ex.message)),
    lambda ex, obj: logger.error("Unknown error during opN on {} with {} - crashing: {}".format(obj.friendly_name, obj.some_attr, ex.message)))

@op1check
def wrapper_op1(obj):
  return obj.method_1()

@opNcheck
def wrapper_opN(obj):
  return some_function(obj.some_attr)

忽略空白行，这比您的原始代码更紧凑 10 行，但以 tryblock 和 logged_call 的沉没成本为代价实现；它现在是否更可读是一个见仁见智的问题。

您还可以选择在单独的模块中定义 logged_call 本身以及从它派生的所有不同装饰器，如果这对您的代码有意义的话；因此多次使用每个派生装饰器。

您可能还会发现更多逻辑结构，您可以通过调整实际检查将其纳入 logged_call 中。

但在最坏的情况下，如果每张支票都具有其他支票所没有的逻辑，您可能会发现像您已经写的那样写出每张支票更具可读性。这真的取决于。

为了完整起见，下面是 tryblock 函数的单元测试:

import examplemodule as ex
from unittest import TestCase
class TestTryblock(TestCase):
  def test_tryblock(self):
    def tryf(a, b):
      if a % 2 == 0:
        raise ValueError
      return a + b
    def proc_ve(ex, a, b):
      self.assertIsInstance(ex, ValueError)
      if a % 3 == 0:
        raise ValueError
      return a + b + 10
    def elsef(a, b):
      return a + b + 20
    def finallyf(a, b):
      return a + b + 30
    block = ex.tryblock(tryf, (ValueError, proc_ve))
    self.assertRaises(ValueError, block, 0, 4)
    self.assertRaises(ValueError, block, 6, 4)
    self.assertEqual([5, 16, 7, 18, 9], map(lambda v: block(v, 4), xrange(1, 6)))
    block = ex.tryblock(tryf, (ValueError, proc_ve), elsef=elsef)
    self.assertEqual([25, 16, 27, 18, 29], map(lambda v: block(v, 4), xrange(1, 6)))
    block = ex.tryblock(tryf, (ValueError, proc_ve), elsef=elsef, returnbody=True)
    self.assertEqual([5, 16, 7, 18, 9], map(lambda v: block(v, 4), xrange(1, 6)))
    block = ex.tryblock(tryf, (ValueError, proc_ve), finallyf=finallyf)
    self.assertEqual([5, 16, 7, 18, 9], map(lambda v: block(v, 4), xrange(1, 6)))
    block = ex.tryblock(tryf, (ValueError, proc_ve), finallyf=finallyf, returnfinally=True)
    self.assertEqual([35, 36, 37, 38, 39], map(lambda v: block(v, 4), xrange(1, 6)))