KaioRetry Internals

The `kaioretry.Retry` class

Not considering the 2 main functions kaioretry.retry() and kaioretry.aioretry() decorators, KaioRetry is basically split in two main classes, with shared responsabilities:

Retry is in charge of handling the decoration and the retry process;
and Context is in charge of keeping track of the try count and delaying management.

The object can be used to decorate either regular or coroutine function.

from kaioretry import Retry

retry = Retry(ValueError)

@retry
def something_we_dont_wanna_see_crashing():
    ...

@retry
async def some_async_thing_we_dont_wanna_see_crashing():
    # The decorated version of this function will still be async
    ...

You can also use the object methods to explicitly decorate as regular or coroutine function.

from kaioretry import Retry

retry = Retry(ValueError)

# On one hand, the retry method produces regular functions.
@retry.retry
def something_we_dont_wanna_see_crashing():
    ...

# On the other hand, the aioretry method produces coroutine functions.
@retry.aioretry
async def some_async_thing_we_dont_wanna_see_crashing():
    # The decorated version of this function will still be async
    ...

# From a KaioRetry point of view, you can use aioretry to decorate a
# regular function to produce a coroutine function. It's designed to work.
@retry.aioretry
def something_regular():
    ...

To refine the number of tries, or delay between said tries. You must use a Context object and give it to Retry constructor.

Use and pass a Context object

from kaioretry import Context, Retry

context = Context(tries=3, delay=1)

@Retry(ValueError, context=context)
def genkidama(...):
    ...

Check out the classes documentation and attributes for more fine tuning.

class kaioretry.decorator.Retry(exceptions=<class 'Exception'>, context=<kaioretry.context.Context object>, *, logger=<Logger kaioretry.decorator (WARNING)>)

Objects of the Retry class are retry decorators.

They can decorate both functions and coroutine functions. Every time the decorated function is called and raises an error, it will automatically be retried until the number of tries is exhausted.

Functions can either be decorated by the retry method, the aioretry method, or by the object itself. If the object sed as decorator, an heuristic will attempt to determine what is the best alternative (retry or aioretry), depending of the nature of the function and the context of a event loop.

Parameters:

exceptions (tuple[type[BaseException], ...] | type[BaseException]) – Exception classes that will trigger another try. Other exceptions raised by the decorated function will not trigger a retry. The value of the exceptions parameters can be either an Exception class or a tuple of Exception classes or whatever is suitable for an except clause. The default is the Exception class, which means any error will trigger a new try.
context (Context) – a Context object that will be used to maintain try count and the delay between them. If omitted, a Context with an infinite nunmber of tries and no delay betwen them will be used.
logger (Logger) – the logging.Logger to which the log messages will be sent to.

__call__(func)

Decorate a function the most accurately possible.

Return type:: Callable[[ParamSpec(FuncParam)], TypeVar(FuncRetVal)]

if func is a function that involves asyncio, use: aioretry(), else use retry().

DEFAULT_LOGGER: Final[Logger] = <Logger kaioretry.decorator (WARNING)>: The logging.Logger object that will be used if none are provided to the constructor.

DEFAULT_CONTEXT: Final[Context] = <kaioretry.context.Context object>

A default Context that will be used if none are provided to the constructor.

It will provide an infinity of tries with no delay between them.

retry(func)

This method is a decorator. The returned and newly-produced function will the same signature, docstring and type annotations as the original one but will also transparently be able to retry when an exception is raised, as described earlier.

If you intend to obtain retry mechanism on an asyncio-compatible coroutine function, look at the aioretry() instead.

Parameters:: func (Callable[[ParamSpec(FuncParam)], TypeVar(FuncRetVal)]) – Any function. Really.
Return type:: Callable[[ParamSpec(FuncParam)], TypeVar(FuncRetVal)]
Returns:: A same-style function.

aioretry(func)

Similar to retry(), this method is a decorator and will produce exact the same result, except that the decorated function is a Coroutine, and that delays induced by the delay constructor parameter and its friends, will be implemented with asyncio functions.

That means the decorated version of the function will be eligible to asyncio.run() or to an await statement, even if given func parameter is not originally an async function to begin with.

Parameters:: func (Callable[[ParamSpec(FuncParam)], Awaitable[TypeVar(FuncRetVal)]] | Callable[[ParamSpec(FuncParam)], TypeVar(FuncRetVal)]) – any callable. Just told you.
Return type:: Callable[[ParamSpec(FuncParam)], Coroutine[None, None, TypeVar(FuncRetVal)]]
Returns:: an async function that will return the same result as the original function’s once awaited.

classmethod is_func_async(func)

Tell if a function can be considered async, either because it’s a Coroutine, an AsyncGenerator or because it is annotated to return collections.abc.Awaitable or typing.Awaitable.

Parameters:: func (Callable[[ParamSpec(FuncParam)], Any]) – any callable, basically.
Return type:: TypeGuard[Callable[[ParamSpec(FuncParam)], Awaitable[Any]]]

The `kaioretry.Context` class

A Retry Context is the time keeper and an accountant: its responsible for maintaining the delay value and the retry count.

Limited number of tries (iterations)

>>> from kaioretry import Context
>>> context = Context(tries=4)
>>> for i, _ in enumerate(context):
...    print(i)
...
0
1
2
3
>>>

Should you wish to persist indefinitely. It is supported.

Unlimited number of tries (iterations)

>>> from kaioretry import Context
>>> context = Context(tries=-1)
>>> for i, _ in enumerate(context):
...    print(i)
...
0
1
2
... there it goes
30
31...
.... and so on
25000
.................
# It never stops.

It’s possible to insert delay between tries.

Adding a one second delay between tries.

>>> from kaioretry import Context
>>> context = Context(tries=4, delay=1)
>>> for i, _ in enumerate(context):
...     print(time())
...
1677350589.3510618
1677350590.3776977
1677350591.403194
1677350592.429291
>>>

It will also log its actions and will help keep things being traceable by adding a per-loop identifier to the logs. e.g:

logging loops

>>> import sys, logging
>>> logging.basicConfig(stream=sys.stdout, encoding='utf-8', level=logging.DEBUG)
>>> from kaioretry import Context
>>> for _ in context: pass
...
INFO:kaioretry.context:00cc19af-7339-442f-9804-16eb10788068: 2 tries remaining
INFO:kaioretry.context:00cc19af-7339-442f-9804-16eb10788068: sleeping 0 seconds
INFO:kaioretry.context:00cc19af-7339-442f-9804-16eb10788068: 1 tries remaining
INFO:kaioretry.context:00cc19af-7339-442f-9804-16eb10788068: sleeping 0 seconds
>>> for _ in context: pass
...
INFO:kaioretry.context:1c4ddbdb-f2b0-4377-a840-92ea8c651ac1: 2 tries remaining
INFO:kaioretry.context:1c4ddbdb-f2b0-4377-a840-92ea8c651ac1: sleeping 0 seconds
INFO:kaioretry.context:1c4ddbdb-f2b0-4377-a840-92ea8c651ac1: 1 tries remaining
INFO:kaioretry.context:1c4ddbdb-f2b0-4377-a840-92ea8c651ac1: sleeping 0 seconds
>>>

If you consider this from Retry point of view, it means that you can keep track of calls, delays and number of tries per calls.

If you want to do more fine tuning to delays and tries, check the documentation below.

class kaioretry.context.Context(tries=-1, delay=0, *, update_delay=<function Context.<lambda>>, max_delay=None, min_delay=0, logger=<Logger kaioretry.context (WARNING)>)

The Retry Context will maintain the number of tries and the delay between those tries.

It can act as both a Generator and an AsyncGenerator, and can be reused, multiple times, even with multiple Retry instances.

The Retry objects will iterate over Context, synchronously, or asynchronously, depending of the nature of the decorated function.

Parameters:

tries (int) – the maximum number of iterations (a.k.a.: tries, function calls) to perform before exhaustion. A negative value means infinite. 0 is forbidden, since it would mean “don’t run” at all.
delay (int | float) – the initial number of seconds to wait between two iterations. It must be non-negative. Default is 0 (no delay).
update_delay (Callable[[int | float], int | float]) – a function that will produce the next value of delay value. Can be anything as long as it produces a positive number when called.
max_delay (UnionType[int, float, None]) – the maximum value allowed for delay. If None (the default), then delay is unlimited. Cannot be negative.
min_delay (int | float) – the minimum value allowed for delay. Cannot be negative. Default is 0.
logger (Logger) – the logging.Logger object to which the log messages will be sent to.

Raises:

ValueError – if tries, min_delay or max_delay have incorrect values.

__iter__()

Returns a generator that perform sleep (using regular time.sleep()) between iterations in order to induce delay as instructed.

Return type:: Generator[None, None, None]

async __aiter__()

Returns a asynchronous generator that perform sleep through asyncio.sleep() between iterations in order to induce delay as instructed.

Return type:: AsyncGenerator[None, None]

DEFAULT_LOGGER: Final[Logger] = <Logger kaioretry.context (WARNING)>: The logging.Logger object that will be used if none are provided to the constructor.

Misc Types

Kaioretry helper types

class kaioretry.types.AioretryProtocol(*args, **kwargs)

The typing.Protocol describing the behaviour of the aioretry() decorator.

__call__(func)

When called (to decorate a function), an aioretry decorator will…

Parameters:

func (Callable[[ParamSpec(FuncParam)], Awaitable[TypeVar(FuncRetVal)]] | Callable[[ParamSpec(FuncParam)], TypeVar(FuncRetVal)]) – … take a function as input…

Return type:

Callable[[ParamSpec(FuncParam)], Coroutine[None, None, TypeVar(FuncRetVal)]]

Returns:

… and,

if func returns an Awaitable, then return a same-signature same-type coroutine function.
If func does not, then return a same-signature coroutine function that, once awaited, return func original return value.

KaioRetry Internals

The kaioretry.Retry class

The kaioretry.Context class

Misc Types

The `kaioretry.Retry` class

The `kaioretry.Context` class