The problem
You need to limit function execution time to prevent hanging operations.
The solution
import functools
import signal
from typing import Any, Callable
from concurrent.futures import ThreadPoolExecutor, TimeoutError
class TimeoutException(Exception):
"""Exception raised when a function times out."""
pass
def timeout(seconds: int = 30) -> Callable:
"""Decorator to add timeout to function execution."""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
with ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(func, *args, **kwargs)
try:
return future.result(timeout=seconds)
except TimeoutError:
future.cancel()
raise TimeoutException(
f"Function '{func.__name__}' timed out after {seconds} seconds"
)
return wrapper
return decoratorParameters
secondsintMaximum execution time in seconds
Put it to work
@timeout(seconds=5)
def long_running_operation():
import time
time.sleep(10) # This will timeout
return "Completed"
try:
result = long_running_operation()
except TimeoutException as e:
print(f"Operation timed out: {e}")