Created
August 9, 2018 08:15
-
-
Save rmerkushin/b81b01b2a600eed6b92de55eb9d89b04 to your computer and use it in GitHub Desktop.
Python 3.6+ wait until helper
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import time | |
import requests | |
class TimeoutException(Exception): | |
def __init__(self, message): | |
self.message = message | |
def __str__(self): | |
return repr(self.message) | |
def wait_until(callback, timeout=10, poll_frequency=0.5, message=None): | |
end_time = time.time() + timeout | |
while True: | |
try: | |
result = callback() | |
except Exception: | |
result = False | |
if result: | |
return result | |
time.sleep(poll_frequency) | |
if time.time() > end_time: | |
break | |
if message is None: | |
condition_name = callback.__class__.__name__ | |
message = f'The condition {condition_name} is not reached within {timeout} second(s)' | |
raise TimeoutException(message) | |
# Callback example: | |
class status_code_is_200: | |
def __init__(self, url): | |
self.url = url | |
def __call__(self): | |
response = requests.get(self.url) | |
if response.status_code == 200: | |
return response | |
else: | |
return False | |
# Usage example: | |
result = wait_until( | |
status_code_is_200('https://google.com/'), | |
timeout=5, | |
poll_frequency=0.1, | |
message='resource is not available!' | |
) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment