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.

Medium Python 3.9+ Aug 9, 2026 Errors & debugging 13 views 0 copies

Python code

70 lines
Python 3.9+
import 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

stdout
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

  1. Use `signal.alarm` with `SIGALRM` on Unix-like systems for a more efficient timeout
  2. 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

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Errors & debugging

Related tutorials and quizzes for this topic.