Skip to content

Instantly share code, notes, and snippets.

@thevickypedia
Created May 26, 2022 14:28
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save thevickypedia/03102339347d12609ee74531d7853f3b to your computer and use it in GitHub Desktop.
Save thevickypedia/03102339347d12609ee74531d7853f3b to your computer and use it in GitHub Desktop.
Timeout handler using threading and wrapper
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