-
-
Notifications
You must be signed in to change notification settings - Fork 303
Open
Description
PROBLEM:
Trying to generate success message only if retry got triggered.
Able to generate the warning properly, but not success if retry happened
Using the following code:
import random
import logging
from tenacity import retry, wait_fixed, stop_after_attempt, before_sleep, RetryCallState, retry_if_exception_type
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def my_before_sleep(retry_state: RetryCallState):
if retry_state.outcome.failed:
# If it failed, don't do anything special here; before_sleep handles warnings.
logging.warning(
f"Retrying '{retry_state.fn.__name__}' for the {retry_state.attempt_number} time "
f"after {retry_state.outcome.exception()}..."
)
return retry_state.outcome.result # Return the exception for tenacity to handle
else:
# THIS ELSE IS NOT WORKING, AS EXPECTED
logging.info(
f"'{retry_state.fn.__name__}' completed successfully after retries "
f"(attempt {retry_state.attempt_number}). Result: {retry_state.outcome.result}"
)
def my_after_execute(retry_state: RetryCallState):
"""
Callback function executed after each attempt, regardless of success or failure.
Checks if the function succeeded AND if retries previously occurred.
"""
if not retry_state.outcome.failed:
# IF THE ABOVE ELSE IS NOT SUPPOSED TO WORK, EVEN THIS IF CONDITION IS NOT WORKING.
logging.info(
f"'{retry_state.fn.__name__}' completed successfully after retries "
f"(attempt {retry_state.attempt_number}). Result: {retry_state.outcome.result}"
)
return retry_state.outcome.result # Return the actual result for tenacity to propagate
@retry(
wait=wait_fixed(1),
stop=stop_after_attempt(3),
retry=retry_if_exception_type(ValueError),
before_sleep=my_before_sleep,
after=my_after_execute,
reraise=True
)
def might_fail_function(prob):
# Simulate success or failure
current_prob = random.random()
if current_prob < prob:
logging.info("Function failed on this attempt.")
logging.info(f"current prob: {current_prob}")
raise ValueError("Simulated failure")
else:
logging.info("Function succeeded on this attempt.")
logging.info(f"current prob: {current_prob}")
return "Operation Completed!"
logging.info("Attempt 1: Should fail initially and then succeed with retries")
# fail the function in 1st try and then succeed (THIS PART IS NOT WORKING)
might_fail_function(0.5)
logging.info('-' * 50)
LOG:
Expecting the log to generate success message either through custom after callback, or through custom before_sleep callback. But none of that is working, as you can see the simulation Attempt 1:
Collecting tenacity
Downloading tenacity-9.1.2-py3-none-any.whl.metadata (1.2 kB)
Downloading tenacity-9.1.2-py3-none-any.whl (28 kB)
Installing collected packages: tenacity
Successfully installed tenacity-9.1.2
[notice] A new release of pip is available: 24.2 -> 25.1.1
[notice] To update, run: python.exe -m pip install --upgrade pip
2025-06-29 12:45:17,638 - INFO - Attempt 1: Should fail initially and then succeed with retries
2025-06-29 12:45:17,638 - INFO - Function failed on this attempt.
2025-06-29 12:45:17,640 - INFO - current prob: 0.16599903411725936
2025-06-29 12:45:17,641 - WARNING - Retrying 'might_fail_function' for the 1 time after Simulated failure...
2025-06-29 12:45:18,643 - INFO - Function succeeded on this attempt.
2025-06-29 12:45:18,643 - INFO - current prob: 0.7810787153522702
Metadata
Metadata
Assignees
Labels
No labels