如何用Python装饰器优化REST API请求重试逻辑以避免代码重复

我有多个执行REST API请求的函数，为处理错误场景，我写了这样的重试代码：

no_of_retries = 3 
def check_status(): 
    for i in range(0,no_of_retries): 
        url = "http://something/something" 
        try: 
            result = requests.get(url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json() 
            if 'error' not in result: 
                return result 
            else: 
                continue 
        except Exception as e: 
            continue 
    return None 

还有很多类似操作的方法，请问怎么用装饰器优化代码，避免重复逻辑？

这场景简直是装饰器的完美用武之地！我们可以把重复的重试循环、异常捕获、结果校验逻辑全部封装成一个装饰器，让你的每个API请求函数只需要专注于核心的请求逻辑，彻底告别代码冗余。

第一步：编写通用的重试装饰器 #

我们先写一个可配置、可扩展的重试装饰器，它支持自定义重试次数，还能传入自定义的成功判断规则：

import functools
import requests
from requests.auth import HTTPBasicAuth

def retry_on_failure(max_retries=3, success_check=None):
    def decorator(func):
        # 用functools.wraps保留原函数的元信息，比如函数名、文档字符串
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    # 调用原函数，也就是发起API请求的逻辑
                    result = func(*args, **kwargs)
                    # 如果有自定义的成功检查函数，就用它判断；否则默认无异常即为成功
                    if success_check is None or success_check(result):
                        return result
                    print(f"请求返回结果不符合要求（尝试 {attempt+1}/{max_retries}），重试中...")
                except Exception as e:
                    # 这里可以替换成日志记录，方便后续排查问题
                    print(f"请求抛出异常（尝试 {attempt+1}/{max_retries}）：{str(e)}，重试中...")
            # 所有重试都失败，返回None（也可以根据需求抛出异常）
            print(f"已耗尽所有{max_retries}次重试机会，请求失败")
            return None
        return wrapper
    return decorator

第二步：改造你的check_status函数 #

现在把原来的函数简化，只保留核心的请求代码，然后给它加上装饰器：

# 先定义一个通用的成功检查函数：判断返回的JSON里没有'error'字段
def api_result_has_no_error(result):
    return 'error' not in result

# 给函数加上装饰器，指定重试次数和成功检查规则
@retry_on_failure(max_retries=3, success_check=api_result_has_no_error)
def check_status():
    url = "http://something/something"
    return requests.get(url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()

第三步：复用装饰器到其他API函数 #

其他类似的请求函数可以直接复用这个装饰器，甚至可以根据需求调整参数，比如给某个函数设置更少的重试次数：

@retry_on_failure(max_retries=2, success_check=api_result_has_no_error)
def get_user_details(user_id):
    url = f"http://something/users/{user_id}"
    return requests.get(url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()

这么做的好处 #

• 彻底消除重复代码：所有重试、异常处理逻辑都集中在装饰器里，不用每个函数都写一遍循环
• 灵活性拉满：不同函数可以设置不同的重试次数，甚至不同的成功判断规则（比如有的接口需要判断code == 200）
• 职责更清晰：每个API函数只需要负责发起请求，重试的事情交给装饰器处理
• 维护成本更低：以后如果要修改重试逻辑（比如添加重试延迟、只捕获特定异常），只需要修改装饰器这一处即可

内容的提问来源于stack exchange，提问作者PGS