python - 使用 Python 装饰器处理 try 和 exception

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。

硬编码如下所示，但是向每个协议(protocol)添加相同的逻辑是很愚蠢的。

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