hellysmile / async_retrying Goto Github PK

https://docs.python.org/3.10/library/asyncio-task.html#generator-based-coroutines

Generally speaking the asyncio.coroutine decorator and yield from is no longer supported in Python 3.11 version of asyncio.

https://github.com/wikibusiness/async_retrying/blob/master/async_retrying.py#L130

python==3.10
async_retrying==0.2.2

error:
TypeError: sleep() got an unexpected keyword argument 'loop'
This may be caused by the removal of loop parameter in later versions of python. Can you optimize it?

Source place:

@asyncio.coroutine
def callback(attempt, exc, args, kwargs, delay=None, *, loop):
  if delay is None:
      delay = callback.delay
yield from asyncio.sleep(attempt * delay, loop=loop)
return retry

Right now if is in debug mode, no attempts is tried

It is impossible to use kwargs = True and immutable = True because impossible to make copy.deepcopy from loop.

I dont want to get logging massage, could you add a params like log=False to close logger.warning msg。
if log: logger.warning( exc.__class__.__name__ + ' -> Attempts (%(attempt)d) are over for %(fn)r', # noqa context, exc_info=exc,

My proposal is to make retry function possible to be used as context manager which wraps original function in the with block:

>>> from asynctest import CoroutineMock
>>> async def case1():
...     fn = CoroutineMock(side_effect=[RuntimeError, 'success'])
...     return await fn()
>>> async def case2():
...     fn = CoroutineMock(side_effect=[RuntimeError, 'success'])
...     with retry(fn) as wrapped:
...         return await wrapped() 
>>> asyncio.get_event_loop().run_until_complete(case1())
RuntimeError:
...
>>> asyncio.get_event_loop().run_until_complete(case2())
'success'