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)