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

py49o6xq  于 2024-01-05  发布在  Python
关注(0)|答案(2)|浏览(184)

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

  1. def retry(f) -> Any:
  2. @contextmanager
  3. def retry_context(*args):
  4. # response = False
  5. i = 0
  6. response_ok = False
  7. retries = DEFAULT_RETRIES
  8. status_retries = DEFAULT_STATUS_RETRIES
  9. retry_count = 0
  10. while i < retries and response_ok is False:
  11. retry_count = retry_count+1
  12. yield response
  13. i += 1
  14. if response.status_code not in status_retries:
  15. response_ok = True
  16. logger.debug("Response {}".format(response))
  17. if not response_ok:
  18. logger.warning("Response ERROR: {}".format(response))
  19. return response
  20. def wrapper(*args, **kwargs) -> Any:
  21. with retry_context(*args):
  22. return f(*args, **kwargs)
  23. async def async_wrapper(*args, **kwargs) -> Any:
  24. with retry_context(*args):
  25. return await f(*args, **kwargs)
  26. if asyncio.iscoroutinefunction(f):
  27. return async_wrapper
  28. return wrapper

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

zlwx9yxi

zlwx9yxi1#

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

5anewei6

5anewei62#

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

  1. from tenacity import retry, retry_if_result, stop_after_attempt
  2. retries = 3
  3. status_retries = (500, 502)
  4. def is_retryable_status_code(response):
  5. global status_retries
  6. return response.status_code in status_retries
  7. def retry_with_status(f) -> Any:
  8. global retries
  9. @retry(stop = stop_after_attempt(retries), retry=retry_if_result(is_retryable_status_code))
  10. def retry_wrapper(*args, **kwargs):
  11. return f(*args, **kwargs)
  12. @retry(stop = stop_after_attempt(retries), retry=retry_if_result(is_retryable_status_code))
  13. async def async_retry_wrapper(*args, **kwargs):
  14. return await f(*args, **kwargs)
  15. def wrapper(*args, **kwargs) -> Any:
  16. global retries, status_retries
  17. if 'retry' in dir(args[0]) and 'status_retries' in dir(args[0]):
  18. retries = args[0].retries
  19. status_retries = args[0].status_retries
  20. if asyncio.iscoroutinefunction(f):
  21. return async_retry_wrapper(*args, **kwargs)
  22. else:
  23. return retry_wrapper(*args, **kwargs)
  24. return wrapper

字符串

展开查看全部

相关问题