装饰器如何统一处理函数异常?实现重试机制的代码模板是怎样的?

装饰器通过捕获异常并分别处理不同类型的错误来提升代码的健壮性和可维护性。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来缓存装饰器的结果,避免重复计算。此外,如果装饰器之间存在依赖关系,可以考虑将它们合并成一个装饰器。

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

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

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

© 版权声明
THE END
喜欢就支持一下吧
点赞9 分享