How to Simulate Timeout with Custom TimeoutError in Python
Run a function in a daemon thread and raise a custom TimeoutError if it exceeds a specified time limit.
Python code
70 linesimport time
from typing import Callable, TypeVar
T = TypeVar("T")
class TimeoutError(Exception):
"""Raised when an operation exceeds its time limit."""
def __init__(self, message: str = "Operation timed out"):
self.message = message
super().__init__(self.message)
def run_with_timeout(func: Callable[..., T], timeout_seconds: float, *args, **kwargs) -> T:
"""
Execute a function with a timeout constraint.
Args:
func: The callable to execute
timeout_seconds: Maximum number of seconds allowed
*args, **kwargs: Arguments passed to func
Returns:
The result of func
Raises:
TimeoutError: If the function takes longer than timeout_seconds
RuntimeError: If the function raises an exception
"""
start_time = time.monotonic()
result = None
error = None
def wrapper():
nonlocal result, error
try:
result = func(*args, **kwargs)
except Exception as exc:
error = exc
import threading
thread = threading.Thread(target=wrapper, daemon=True)
thread.start()
thread.join(timeout_seconds)
if thread.is_alive():
raise TimeoutError(f"Operation exceeded {timeout_seconds} seconds")
if error:
raise RuntimeError(f"Function raised: {error}") from error
return result
def slow_task(delay: float) -> str:
"""Simulate a task that takes time."""
time.sleep(delay)
return f"Completed after {delay} seconds"
if __name__ == "__main__":
# Example: quick task succeeds
print(run_with_timeout(slow_task, 3, delay=1))
# Example: slow task triggers timeout
try:
run_with_timeout(slow_task, 0.5, delay=2)
except TimeoutError as e:
print(f"Timeout: {e.message}")
Output
Completed after 1 seconds
Timeout: Operation exceeded 0.5 seconds
How it works
The run_with_timeout function starts the target in a separate daemon thread and joins it with a timeout. thread.join(timeout_seconds) blocks until the thread finishes or the timeout elapses. If the thread is still alive after joining, a custom TimeoutError is raised. The nonlocal keyword lets the wrapper assign results and errors back to the outer scope. Any exception from the function is re-raised as RuntimeError to distinguish it from timeouts.
Common mistakes
- Using `time.sleep` instead of `time.monotonic` for duration checks, which can be affected by system clock changes
- Forgetting to set `daemon=True`, causing the program to hang when the main thread exits
- Raising the original exception directly instead of wrapping it in `RuntimeError`, losing context
Variations
- Use `signal.alarm` with `SIGALRM` on Unix-like systems for a more efficient timeout
- Wrap the function in a `concurrent.futures.ThreadPoolExecutor` and use `future.result(timeout=N)`
Real-world use cases
- Calling external APIs with strict response-time SLAs in a data ingestion pipeline
- Enforcing time bounds on database queries in a web service to prevent slow-query pileups
- Adding a watchdog to crawling or scraping jobs so a single stuck page doesn't stall the whole batch
Sponsored
More from Errors & debugging
- Catch RecursionError and Fail Gracefully in Python easy
- Catch ValueError and print friendly message in Python easy
- Collect Multiple Validation Errors in Python Before Raising medium
- Handle ValueError and ZeroDivisionError in Python with try except easy
- How to Add a Correlation ID to Logging Records in Python medium
- How to Assert Preconditions with Descriptive Messages in Python easy
Keep learning
Related tutorials and quizzes for this topic.