使用Python装饰器处理尝试和异常

Question

如何以装饰器的方式传递try函数和异常处理程序函数

例如，这是WCF连接。

def wcf():
    def send_request(result):
        # establish connection...
        result["response"] = True

    def handle_execption(e, result):
        # logger something
        result["error"] = str(e)
    result = {}
    try:
        send_request(result)
    except Exception as e:
        handle_execption(e, result)

现在，我想在这个连接中添加一个重试机制。实现这个目标的最佳方法是什么，我有多种连接方式(比如REST、SOAP WCF等)？通常，它们共享相同的模式，并且都有send_request和handle_execption。

硬代码之一如下所示，但是在每个协议中添加相同的逻辑是非常愚蠢的。

for attempt in range(0, 3):
    sleep_seconds = attempt ** 2
    sleep(sleep_seconds)

    try:
        send_request(result)
        break
    except Exception as e:
        handle_execption(e, result)

Answer 1

如果您总是以相同的方式处理异常，则可以编写一个装饰器来对该行为进行硬编码并处理重试逻辑：

def retry_decorator(base_function):
    def new_function(*args, **kwargs):  # This allows you to decorate functions without worrying about what arguments they take
        for attempt in range(3):
            sleep_seconds = attempt ** 2
            sleep(sleep_seconds)

            try:
                return base_function(*args, **kwargs)  # base_function is whatever function this decorator is applied to
            except Exception as e:
                print(e)  # Replace this with whatever you want, as long as it's the same for all possible base_functions

    return new_function


@retry_decorator  # Replaces send_request() with retry_decorator(send_request)()
def send_request(result):
    result["response"] = True

如果您想对不同的连接使用不同的异常处理逻辑，事情就会变得更加复杂：

def retry_decorator_2(exception_handle_function):
    def new_decorator(base_function):
        def new_function(*args, **kwargs):
            for attempt in range(3):
                sleep_seconds = attempt ** 2
                sleep(sleep_seconds)

                try:
                    return base_function(*args, **kwargs)
                except Exception as e:
                    exception_handle_function(e)

        return new_function

    return new_decorator


@retry_decorator_2(print)  # Replace 'print' with an exception handling function
def send_request(result):
    result["response"] = True

在本例中，retry_decorator_2(print)创建一个新的装饰器(它将print存储为exception_handle_function)，然后将该装饰器应用于send_request函数。

如果需要exception_handle_function接受参数，可以将它们作为装饰器的一部分传递：

def retry_decorator_3(exception_handle_function, *exception_function_args, **exception_function_kwargs):
    def new_decorator(base_function):
        def new_function(*args, **kwargs):
            for attempt in range(3):
                sleep_seconds = attempt
                sleep(sleep_seconds)

                try:
                    return base_function(*args, **kwargs)
                except Exception as e:
                    exception_handle_function(e, *exception_function_args, **exception_function_kwargs)

        return new_function

    return new_decorator


@retry_decorator_3(print, 1, 2, 3)  # 1, 2, 3 become arguments for the print function, or whatever you replace it with
def send_request(result):
    result["response"] = True

Answer 2

为什么是装饰师？我觉得它和

def wcf():
    def send_request(result):
        # establish connection...
        result["response"] = True

    def handle_exception(e, result):
        # logger something
        result["error"] = str(e)

    return retry(send_request, handle_exception)


def retry(send_request, handle_exception):
    result = {}

    for attempt in range(0, 3):
        sleep(attempt ** 2)

        try:
            send_request(result)
            return result

        except Exception as e:
            return handle_execption(e, result)