python 如何将收益率值分配给变量

py49o6xq 于 12个月前发布在 Python

关注(0)|答案(2)|浏览(147)

我试图为sync和declare函数创建一个动态装饰器函数，它工作得很好，但我不能进一步使用yield value。

def retry(f) -> Any:
    @contextmanager
    def retry_context(*args):
        # response = False
        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 = retry_count+1
            yield response
            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):
            return f(*args, **kwargs)

    async def async_wrapper(*args, **kwargs) -> Any:
        with retry_context(*args):
            return await f(*args, **kwargs)

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

字符串
我已经看到了许多其他的方法来给yield赋值，但没有一个能解决我的问题，因为我还想进一步使用它。任何其他的解决方案也是受欢迎的。

python

来源：https://stackoverflow.com/questions/77743124/how-to-assign-yield-value-to-variable

2条答案

按热度按时间

zlwx9yxi1#

上下文管理器协议（即with语句及其相关的特殊方法和contextlib模块中的帮助函数），或者在本例中也是“context manager”，不是设计成并且不能使其本身作为“重试”机制工作。
这意味着：上下文管理器，或者是用contextlib.contextmanager装饰器装饰的生成器，或者是具有__enter__和__exit__方法的成熟类，将只，并且只能通过语言的工作方式，将控制传递给with语句的主体一次。对于contextmanager装饰的生成器，这意味着它必须在其生命周期中运行yield表达式一次。
您应该使用外部while循环和不同设计的类，或者使用现成的重试库来实现您的目标。
至于一个很好的第三方库，它实现了很多选项的重试，并且关心边缘情况，你可以尝试“tenacity”：https://github.com/jd/tenacity

赞(0）回复(0）举报 12个月前

5anewei62#

我很有毅力。谢谢大家的指导。

from tenacity import retry, retry_if_result, stop_after_attempt

retries = 3  
status_retries = (500, 502)

def is_retryable_status_code(response):
    global status_retries
    return response.status_code in status_retries

def retry_with_status(f) -> Any:
    global retries
    @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

字符串

赞(0）回复(0）举报 12个月前

我来回答

python 如何将收益率值分配给变量

2条答案

相关问题

热门标签

最新问答