Overwrite retry_hooks on function or at call location.

[StefanBRas]

I have cases where I would like to log retries for some functions in one way and for other functions in another way.

It looks to me like this can only be done by changing the global retry hooks between calls. I'm also not sure how that would work with threading or async.

Concretely I have cases where I expect an error and I don't want to log those as warnings. This often happen if I'm waiting for some service to come online, then I expect the first call to fail.

Maybe it could be allow to specify some retry_hook args when stamina.retry that would then be set on the RetryDetails and passed to the retry_hook:

import stamina
import random
import logging

from stamina.instrumentation import RetryDetails, RetryHook, RetryHookFactory


def my_retry_hook() -> RetryHook:
    def log_retries(details: RetryDetails) -> None:
        logging.getLogger("stamina").log(
            logging.WARNING if details.retry_args.get("important") else logging.DEBUG,
            "stamina.retry_scheduled",
        )

    return log_retries


stamina.instrumentation.set_on_retry_hooks([RetryHookFactory(my_retry_hook)])


@stamina.retry(on=ValueError, retry_args={"important": True})
def important():
    if random.random() < 0.6:
        raise ValueError("Big Mistake")
    return "Important Result"


@stamina.retry(on=ValueError)
def not_important():
    if random.random() < 0.6:
        raise ValueError("Small Mistake")
    return "Not important Result"

not_important() # retries logged as debug

important()  # retries logged as warning

[hynek]

Sorry for the late response. Couldn't possibly the name of the function that is part of the details work as such filter? That would have the upside that you wouldn't litter your actual code with introspection-specific gunk. Also that said, is it such a big deal to have a few extra log entries that would warrant such heavy-handed changes?

[StefanBRas]

It could, but I don't think it would scale very nicely. And it would be the same issue as I want to avoid that the configuration of behavior happens separately from where it's used. Also I wouldn't be able to use it with the retry_context blocks.

The extra log issues would be pretty annoying - for a large application I would either spam the logs with warning entries that are actually debug and can be mostly ignored or log actual warning entries as debug, thereby hiding them. I can make log filters, but that goes back to the original problem of not setting the behavior at the call site. Other developers would have to "just" know that the logging behavior is hardcoded for each function and such in a completely different file.

Looking at the code, it doesn't seem like that invasive of a change? And it would be fully backwards compatible as far as I can see. Not to argue that it should be implemented because it's easy, just to say that if it were to be done it would be easy.

And thanks for taking the time to discuss this!

[StefanBRas]

I thought about it some more, and I think using context vars makes more sense for my setup. Then i can have default behavior that I can overwrite for a function and again at the place I use it.

I've put a PoC of the setup below, maybe others will find it useful too

import stamina
from functools import wraps
from contextvars import ContextVar

from stamina.instrumentation import RetryDetails

_retry_hook_log_level = ContextVar("log_level", default=20)
_retry_hook_no_logging = ContextVar("no_logging", default=False)

def my_retry_hook(details: RetryDetails) -> None:
    if _retry_hook_no_logging.get():
        return
    print(
        _retry_hook_log_level.get(),
        "stamina.retry_scheduled",
    )

stamina.instrumentation.set_on_retry_hooks([my_retry_hook])

class stamina_args:
    def __init__(self, log_level: int = 20, no_logging: bool = False):
        self._log_level = log_level
        self._no_logging = no_logging

    def __call__(self, func):  # yes, needs fancy typing here
        @wraps(func)
        def _wrapper(*args, **kwargs):
            reset_token = _retry_hook_log_level.set(self._log_level)
            res = func(*args, **kwargs)
            _retry_hook_log_level.reset(reset_token)
            return res

        return _wrapper

    def __enter__(self):
        self.reset_token_log_level = _retry_hook_log_level.set(self._log_level)
        self.reset_token_no_logging = _retry_hook_no_logging.set(self._no_logging)

    def __exit__(self, exc_type, exc_val, exc_tb):
        _retry_hook_log_level.reset(self.reset_token_log_level)
        _retry_hook_no_logging.reset(self.reset_token_no_logging)


@stamina_args(log_level=10)
@stamina.retry(on=ValueError, attempts=2)
def decorated_function():
    raise ValueError("nope")


def test_decorator():
    decorated_function()  # prints with log level 10


def test_overwrite_with_context_manager():
    with stamina_args(no_logging=True):
        decorated_function()  # Doesn't print

You could also wrap stamina.retry instead, but it's roughly the same