exception_backoff.py

import typing as t
import datetime as dt
from asyncio import sleep
from functools import wraps

from aiologger import Logger

__all__ = ['exception_backoff']


def exception_backoff(
    logger: Logger,
    interval: int,
    exceptions: t.List[t.Type[Exception]],
    msg: t.Optional[str],
) -> t.Callable:
    def decorator(func: t.Callable):
        @wraps(func)
        async def wrapper(*args, **kwargs) -> t.Callable:
            calls_count = 0
            while True:
                if calls_count:
                    await sleep(interval)
                try:
                    result = await func(*args, **kwargs)
                except Exception as err:
                    if not type(err) in exceptions:
                        raise
                    if calls_count:
                        await logger.debug(msg=f"{msg}. This a {calls_count} try.")
                    calls_count += 1
                    await logger.warning(
                        msg=f"Type: {err.__class__.__name__}, msg: {str(err)}. "
                            f"Planed retry: {dt.datetime.utcnow() + dt.timedelta(seconds=interval)}"
                    )
                else:
                    return result

        return wrapper

    return decorator