Created
May 26, 2022 14:28
-
-
Save thevickypedia/03102339347d12609ee74531d7853f3b to your computer and use it in GitHub Desktop.
Timeout handler using threading and wrapper
This file contains 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 functools | |
import time | |
from threading import Thread | |
from typing import Union, Callable, NoReturn | |
def timeout(duration: Union[float, int]) -> Callable: | |
"""Timeout handler for Windows OS. | |
Args: | |
duration: Takes the duration as an argument. | |
Returns: | |
Callable: | |
Returns the decorator method. | |
""" | |
def decorator(func: Callable) -> Callable: | |
"""Decorator for the function to be timed out. | |
Args: | |
func: Function that is to be timed out. | |
Returns: | |
Callable: | |
Returns the wrapper. | |
""" | |
@functools.wraps(wrapped=func) | |
def wrapper(*args, **kwargs) -> TimeoutError: | |
"""Wrapper for the function that has to be timed out. | |
Args: | |
*args: Arguments for the function. | |
**kwargs: Keyword arguments for the function. | |
Raises: | |
TimeoutError: | |
Raises TimeoutError as specified. | |
""" | |
result = {'ERROR': TimeoutError(f'Function [{func.__name__}] exceeded the timeout [{duration} seconds]!')} | |
def place_holder() -> NoReturn: | |
"""A place-holder function to be called by the thread.""" | |
try: | |
result['ERROR'] = func(*args, **kwargs) | |
except TimeoutError as error_: | |
result['ERROR'] = error_ | |
thread = Thread(target=place_holder, daemon=True) | |
try: | |
thread.start() | |
thread.join(timeout=duration) | |
except Exception as error: | |
print("called") | |
raise error | |
base_error = result['ERROR'] | |
if isinstance(base_error, TimeoutError): | |
raise TimeoutError( | |
base_error | |
) | |
return base_error | |
return wrapper | |
return decorator | |
@timeout(duration=3) | |
def sleeper(duration): | |
time.sleep(duration) | |
if __name__ == '__main__': | |
sleeper(duration=2) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment