Python装饰器实现异常重试机制

从现在开始，我们要努力学习啦！今天我给大家带来《装饰器统一处理函数异常并实现重试机制，是一种常见的做法，尤其在需要增强函数健壮性、自动重试失败操作的场景中非常有用。下面是一个通用的代码模板，结合了异常捕获和重试逻辑。✅ 一、基本思路使用 Python 的 functools.wraps 来保留被装饰函数的元数据。使用 try...except 捕获指定的异常类型。在异常发生时，根据设定的重试次数进行重试。可以设置重试间隔时间（可选）。支持自定义重试条件（如只对特定异常重试）。✅ 二、代码模板 import time import functools from typing import Callable, Optional, Any def retry(max_retries: int = 3, delay: float = 1.0, exceptions: tuple = (Exception,)): """ 装饰器：用于对函数添加重试机制 :param max_retries: 最大重试次数 :param delay: 重试间隔时间（秒） :param exceptions: 需要重试的异常类型 """ def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs) -> Any: retries_left =》，感兴趣的朋友请继续看下去吧！下文中的内容我们主要会涉及到等等知识点，如果在阅读本文过程中有遇到不清楚的地方，欢迎留言呀！我们一起讨论，一起学习！

装饰器通过捕获异常并分别处理不同类型的错误来提升代码的健壮性和可维护性。1. 装饰器本质上是语法糖，用于包裹目标函数并在其外部统一处理异常；2. 可以针对不同异常类型编写特定逻辑，例如对网络超时进行重试，而对参数错误直接抛出异常；3. 为避免装饰器嵌套导致性能下降，应简化内部逻辑、使用缓存或合并依赖装饰器；4. 装饰器还可用于权限校验、日志记录、性能监控、数据验证、缓存和事务管理等场景，从而扩展函数行为而不修改其本身。

装饰器本质上是语法糖，利用它可以优雅地统一处理函数异常，顺便还能加上重试机制。这东西用好了，代码瞬间清爽不少。

解决方案

核心思路是，装饰器包裹目标函数，在装饰器内部捕获异常，然后根据需要进行处理，比如记录日志、重试、返回默认值等等。

import functools
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def retry(max_retries=3, delay=1, logger=None):
    """
    一个通用的重试装饰器，可以自定义重试次数、延迟时间和日志记录器。
    """
    def decorator_retry(func):
        @functools.wraps(func)
        def wrapper_retry(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    retries += 1
                    log_message = f"Function {func.__name__} failed with error: {e}, retrying ({retries}/{max_retries})..."
                    if logger:
                        logger.warning(log_message)
                    else:
                        logging.warning(log_message)
                    time.sleep(delay)
            # 如果所有重试都失败了，可以选择抛出异常或返回默认值
            raise Exception(f"Function {func.__name__} failed after {max_retries} retries.")

        return wrapper_retry
    return decorator_retry


def exception_handler(default_value=None, logger=None):
    """
    一个简单的异常处理装饰器，可以记录日志并返回默认值。
    """
    def decorator_exception_handler(func):
        @functools.wraps(func)
        def wrapper_exception_handler(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                log_message = f"Function {func.__name__} raised an exception: {e}"
                if logger:
                    logger.error(log_message)
                else:
                    logging.error(log_message)
                return default_value  # 返回默认值
        return wrapper_exception_handler
    return decorator_exception_handler


@retry(max_retries=3, delay=2, logger=logging)  # 使用retry装饰器，最多重试3次，每次间隔2秒
@exception_handler(default_value="Error", logger=logging) # 使用exception_handler装饰器，发生异常时返回"Error"
def might_fail(attempt):
    """
    一个可能失败的函数，用于演示。
    """
    print(f"Attempt {attempt}: Trying to execute might_fail...")
    if attempt < 2:
        raise ValueError(f"Simulated failure on attempt {attempt}")
    return f"Success on attempt {attempt}"


# 示例用法
if __name__ == "__main__":
    for i in range(3):
        try:
            result = might_fail(i)
            print(f"Result: {result}")
            break
        except Exception as e:
            print(f"Main block caught an exception: {e}")

装饰器如何处理不同类型的异常？

装饰器可以针对不同类型的异常进行不同的处理。例如，可以针对网络请求超时进行重试，而对于参数错误则直接抛出异常。

import functools
import time
import logging
import requests

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def specific_exception_handler(func):
    """
    针对特定异常类型的处理装饰器。
    """
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except requests.exceptions.RequestException as e:
            logging.error(f"Network error in {func.__name__}: {e}")
            # 可以选择重试网络请求，或者返回一个特定的错误值
            return None
        except ValueError as e:
            logging.error(f"Value error in {func.__name__}: {e}")
            raise  # 重新抛出ValueError，因为这通常是输入错误，不应该被忽略
        except Exception as e:
            logging.error(f"Unexpected error in {func.__name__}: {e}")
            return None  # 对于其他未知的异常，返回None
    return wrapper

@specific_exception_handler
def fetch_data(url):
    """
    模拟获取数据的函数，可能会抛出网络请求异常或ValueError。
    """
    print(f"Fetching data from {url}...")
    response = requests.get(url, timeout=5)
    response.raise_for_status()  # 如果响应状态码不是200，则抛出HTTPError
    return response.json()


if __name__ == "__main__":
    try:
        data = fetch_data("https://api.example.com/data")  # 假设这是一个有效的API endpoint
        print(f"Data: {data}")
    except Exception as e:
        print(f"Main block caught an exception: {e}")

    try:
        data = fetch_data("invalid_url")  # 模拟一个无效的URL，会抛出requests.exceptions.RequestException
        print(f"Data: {data}")
    except Exception as e:
        print(f"Main block caught an exception: {e}")

如何避免装饰器过度嵌套导致性能下降？

装饰器嵌套确实可能导致性能问题，尤其是当装饰器内部逻辑比较复杂时。解决这个问题，一方面要尽量简化装饰器内部的逻辑，另一方面可以考虑使用functools.lru_cache来缓存装饰器的结果，避免重复计算。此外，如果装饰器之间存在依赖关系，可以考虑将它们合并成一个装饰器。

除了重试和异常处理，装饰器还能用于哪些场景？

除了重试和异常处理，装饰器还能用于很多其他的场景，比如：

• 权限校验： 检查用户是否有权限访问某个函数。
• 日志记录： 记录函数的调用信息，包括参数和返回值。
• 性能监控： 统计函数的执行时间。
• 数据验证： 验证函数的输入参数是否符合要求。
• 缓存： 缓存函数的返回值，避免重复计算。
• 事务管理： 在函数执行前后开启和关闭数据库事务。

本篇关于《Python装饰器实现异常重试机制》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！