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)
earliest_attempt
that you never make use of? \$\endgroup\$