app/util/retry_deco_async.py
import asyncio
import functools
from typing import Callable
from loguru import logger
def retry_async(exceptions=Exception, tries=3, delay=0) -> Callable[[Callable], Callable]:
def deco_retry(f):
@functools.wraps(f)
async def f_retry(*args, **kwargs):
m_tries, m_delay = tries, delay
while m_tries > 1:
try:
return await f(*args, **kwargs)
except exceptions as e:
logger.warning(f"{e}, Retrying in {m_delay} seconds...")
if m_delay > 0:
await asyncio.sleep(m_delay)
m_tries -= 1
return await f(*args, **kwargs)
return f_retry
return deco_retry
def wrap_object(obj: object, deco: Callable[[Callable], Callable]):
for attr in dir(obj):
if not attr.startswith('_') and asyncio.iscoroutinefunction(attr_val := getattr(obj, attr)):
setattr(obj, attr, deco(attr_val))