如何将yield值赋值给变量？同步异步重试装饰器问题

原代码核心问题 #

1. retry_context中response变量未初始化就使用，会直接触发NameError
2. 普通@contextmanager的yield无法捕获被装饰函数的返回值——with块里直接return f(...)不会把结果传递给yield，上下文管理器的yield主要用于上下文的进入/退出操作，而非捕获函数输出
3. 异步函数使用普通@contextmanager是错误的，必须配合@asynccontextmanager和async with，但即使修正，也很难优雅地捕获返回值做重试判断

推荐解决方案：直接在装饰器Wrapper中处理重试逻辑 #

这种方式无需依赖上下文管理器，直接在同步/异步Wrapper中捕获函数返回值，逻辑更清晰，兼容性更好：

import asyncio
import logging

logger = logging.getLogger(__name__)
DEFAULT_RETRIES = 3
# 定义需要触发重试的HTTP状态码（可根据需求调整）
DEFAULT_STATUS_RETRIES = {500, 502, 503, 504}

def retry(f):
    def sync_wrapper(*args, **kwargs):
        retry_count = 0
        response_ok = False
        last_response = None

        while retry_count < DEFAULT_RETRIES and not response_ok:
            retry_count += 1
            try:
                last_response = f(*args, **kwargs)
                # 判断响应是否符合预期，不符合则继续重试
                if last_response.status_code not in DEFAULT_STATUS_RETRIES:
                    response_ok = True
                    logger.debug(f"请求成功: {last_response}")
                    return last_response
                logger.debug(f"第 {retry_count}/{DEFAULT_RETRIES} 次重试，原因: 状态码 {last_response.status_code}")
            except Exception as e:
                logger.debug(f"第 {retry_count}/{DEFAULT_RETRIES} 次重试，原因: 异常 {str(e)}")
                # 最后一次重试仍失败则抛出异常
                if retry_count == DEFAULT_RETRIES:
                    raise

        logger.warning(f"重试耗尽，最后一次响应: {last_response}")
        return last_response

    async def async_wrapper(*args, **kwargs):
        retry_count = 0
        response_ok = False
        last_response = None

        while retry_count < DEFAULT_RETRIES and not response_ok:
            retry_count += 1
            try:
                last_response = await f(*args, **kwargs)
                if last_response.status_code not in DEFAULT_STATUS_RETRIES:
                    response_ok = True
                    logger.debug(f"请求成功: {last_response}")
                    return last_response
                logger.debug(f"第 {retry_count}/{DEFAULT_RETRIES} 次重试，原因: 状态码 {last_response.status_code}")
            except Exception as e:
                logger.debug(f"第 {retry_count}/{DEFAULT_RETRIES} 次重试，原因: 异常 {str(e)}")
                if retry_count == DEFAULT_RETRIES:
                    raise

        logger.warning(f"重试耗尽，最后一次响应: {last_response}")
        return last_response

    # 根据被装饰函数是否为协程函数，返回对应Wrapper
    return async_wrapper if asyncio.iscoroutinefunction(f) else sync_wrapper

方案优势 #

• 直接捕获函数返回值，无需绕上下文管理器的yield传递逻辑
• 同步/异步逻辑分离，避免异步场景下的上下文兼容性问题
• 增加了异常重试逻辑（可根据需求移除）
• 明确处理重试次数耗尽的情况，输出日志便于排查

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