Function to call a list of callables with retry

Question

This script is designed to be run via windows task scheduler once per day.

All callables passed to this function should only return a bool. The callables are called until either the maximum number of call attempts is reached, or until they return True. If one or more callables returns False, the program will sleep for the alloted time 'attempt_interval', before attempting to calls again to those which have not yet returned True.

Function:

import time
from dateutil.parser import parse
def call_callables(callables: list,
 max_attempts=12,
 earliest_attempt="07:00",
 attempt_interval=600):
 """
 Call each callable until it either returns True or max_attempts is reached
 :param callables: a list of callable functions/methods which return
 either True or False.
 :param earliest_attempt: For the current day, don't attempt list generation
 before this time. This is a target time for the
 first attempt.
 :param max_attempts: The maximum number of calls to each callable
 :param attempt_interval: The number of seconds to wait between calls to each
 callable
 """
 earliest_attempt = parse(earliest_attempt)
 current_time = datetime.datetime.now()
 # track number of attempts for each callable
 attempt_counter = defaultdict(int)
 # track return bool status for each callable
 success_tracker = defaultdict(bool)
 callable_objs = callables
 while callable_objs:
 for callable_obj in callables:
 success_tracker[callable_obj] = callable_obj()
 attempt_counter[callable_obj] += 1
 if (success_tracker[callable_obj] or attempt_counter[callable_obj]
 >= max_attempts):
 callable_objs.remove(callable_obj)
 continue
 # Unsuccessful (False returned by one or more callables) attempt. Retry.
 if callable_objs:
 time.sleep(attempt_interval)
 # return dicts to allow for testing
 return attempt_counter, success_tracker

Test (using pytest-cov; this passed):

import pytest
from unittest.mock import Mock, patch
@patch("time.sleep")
def test_call_callables(sleep):
 mock_true = Mock()
 mock_false = Mock()
 def ret_true():
 return True
 def ret_false():
 return False
 mock_true.call_method = ret_true
 mock_false.call_method = ret_false
 mocks = [mock_true.call_method, mock_false.call_method]
 attempt_tracker, success_tracker = call_callables(callables=mocks,
 max_attempts=10,
 attempt_interval=1)
 assert {ret_true: 1, ret_false: 10} == dict(attempt_tracker)
 assert sleep.call_count == 10
 assert {ret_true: True, ret_false: False} == dict(success_tracker)

• 301_Moved_Permanently: What's with this earliest_attempt that you never make use of?
• Dave: ^ Yes, I'll delete that now I've decided to use the script via a scheduled task.
• 301_Moved_Permanently: Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers .

Answer 1

You are not allowed to remove items from a list while iterating over the list.

>>> a = ["a", "b", "c", "d"]
>>> for b in a:
... print(a,b)
... a.remove(b)
... 
['a', 'b', 'c', 'd'] a
['b', 'c', 'd'] c
>>> 

You should wait to remove the callable_obj from callable_objs until after the for loop completes. Build a list of callable_obj to remove, and bulk remove them at the end. Or use list comprehension and filter out the successful calls:

callable_objs = [ obj for obj in callable_objs if not success_tracker[obj] ]

Answer 2

Original while loop:

while callable_objs:
 for callable_obj in callables:
 success_tracker[callable_obj] = callable_obj()
 attempt_counter[callable_obj] += 1
 if (success_tracker[callable_obj] or attempt_counter[callable_obj]
 >= max_attempts):
 callable_objs.remove(callable_obj)
 continue
 # Unsuccessful (False returned by one or more callables) attempt. Retry.
 if callable_objs:
 time.sleep(attempt_interval)

To avoid modifying callable_objs list while iterating over it(as mentioned in AJNeufeld's answer):

while callable_objs:
 for callable_obj in callable_objs:
 success_tracker[callable_obj] = callable_obj()
 attempt_counter[callable_obj] += 1
 callable_objs = [obj for obj in callable_objs
 if not success_tracker[obj]
 and attempt_counter[obj] < max_attempts]
 # Unsuccessful (False returned by one or more callables) attempt. Retry.
 if callable_objs:
 time.sleep(attempt_interval)