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) 

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)