如何将yield值分配给变量

Question

我正在尝试创建一个适用于同步和异步函数的动态装饰器，并且它工作正常，但是我无法在后续代码中使用yield返回的值。

def retry(f) -> Any:
    @contextmanager
    def retry_context(*args):
        i = 0
        response_ok = False

        retries = DEFAULT_RETRIES
        status_retries = DEFAULT_STATUS_RETRIES

        retry_count = 0
        while i < retries and response_ok is False:
            retry_count += 1
            response = yield  # 这里接收yield表达式的值
            i += 1
            if response.status_code not in status_retries:
                response_ok = True
                logger.debug("Response {}".format(response))
        if not response_ok:
            logger.warning("Response ERROR: {}".format(response))
        return response

    def wrapper(*args, **kwargs) -> Any:
        with retry_context(*args) as response_generator:
            response = next(response_generator)
            while not response_ok:
                response = f(*args, **kwargs)
                response_ok = response_generator.send(response)
            return response

    async def async_wrapper(*args, **kwargs) -> Any:
        async with retry_context(*args) as response_generator:
            response = await response_generator.asend(None)
            while not response_ok:
                response = await f(*args, **kwargs)
                response_ok = await response_generator.asend(response)
            return response

    if asyncio.iscoroutinefunction(f):
        return async_wrapper
    return wrapper

我已经看到许多其他关于如何向yield赋值的答案，但没有一个能解决我的问题，因为我还希望进一步使用这个值。也非常欢迎其他的解决方案。

DigitalRoss · Answer

上下文管理器协议（即带有with语句及其在contextlib模块中关联的特殊方法和辅助函数），或者在这种情况下是“异步上下文管理器”，本身并未设计为且不能直接用作“重试”机制。

这意味着：无论是使用contextlib.contextmanager装饰器装饰的生成器形式的上下文管理器，还是具有__enter__和__exit__方法的完整类，它们都只能并且按照语言运行方式只能将控制权传递给with语句块一次。对于装饰有contextmanager的生成器上下文管理器而言，这意味着在其生命周期中必须且只能执行一次yield表达式。

要实现你的目标，你需要采用包含不同设计的类的外部while循环，或者使用现成的重试库。

至于一个实现丰富选项并妥善处理边缘情况的重试功能的第三方库，你可以试试“tenacity”：https://github.com/jd/tenacity

Marko Topolnik · Answer

我通过使用tenacity库成功实现了重试功能。感谢大家的指导。

from tenacity import retry, retry_if_result, stop_after_attempt

# 设置重试次数和可重试状态码
retries = 3  
status_retries = (500, 502)

# 定义判断HTTP响应状态码是否需要重试的函数
def is_retryable_status_code(response):
    global status_retries
    return response.status_code in status_retries

# 使用装饰器实现带有状态码判断的重试功能
def retry_with_status(f) -> Any:
    # 同步函数的重试包装器
    @retry(stop=stop_after_attempt(retries), retry=retry_if_result(is_retryable_status_code))
    def retry_wrapper(*args, **kwargs):
        return f(*args, **kwargs)

    # 异步函数的重试包装器
    @retry(stop=stop_after_attempt(retries), retry=retry_if_result(is_retryable_status_code))
    async def async_retry_wrapper(*args, **kwargs):
        return await f(*args, **kwargs)

    # 根据调用函数是否为异步决定使用哪种重试包装器
    def wrapper(*args, **kwargs) -> Any:
        global retries, status_retries
        # 如果被装饰的函数所在对象中包含自定义的重试次数和状态码，进行更新
        if 'retry' in dir(args[0]) and 'status_retries' in dir(args[0]):
            retries = args[0].retries
            status_retries = args[0].status_retries
        
        # 判断被装饰函数是否为异步函数，并相应地返回同步或异步重试包装器的结果
        if asyncio.iscoroutinefunction(f):
            return async_retry_wrapper(*args, **kwargs)
        else:
            return retry_wrapper(*args, **kwargs)

    # 返回最终的装饰器函数
    return wrapper