TimeoutSampler API

Complete reference for the TimeoutSampler class — constructor parameters, iteration protocol, exception handling, and return semantics.

from timeout_sampler import TimeoutSampler

Constructor

TimeoutSampler(
    wait_timeout: float,
    sleep: int,
    func: Callable,
    exceptions_dict: dict[type[Exception], list[str]] | None = None,
    print_log: bool = True,
    print_func_log: bool = True,
    print_func_args: bool = True,
    func_args: tuple[Any] | None = None,
    **func_kwargs: Any,
)

Parameters

Note: When exceptions_dict is omitted (or None), it defaults to {Exception: []}, which silently ignores all exceptions raised inside func until the timeout expires. Pass an explicit empty dict {} to re-raise every exception immediately.

Example — Basic Construction

from timeout_sampler import TimeoutSampler

def check_service():
    return {"status": "ready"}

sampler = TimeoutSampler(
    wait_timeout=30,
    sleep=5,
    func=check_service,
)

Example — Passing Arguments to func

import requests
from timeout_sampler import TimeoutSampler

sampler = TimeoutSampler(
    wait_timeout=60,
    sleep=2,
    func=requests.get,
    func_args=("https://api.example.com/health",),
    timeout=5,          # forwarded as requests.get(..., timeout=5)
)

Iteration Protocol

TimeoutSampler implements __iter__. Use it in a for loop. Each iteration calls func(*func_args, **func_kwargs) and yields the return value.

def __iter__(self) -> Any

Yields: The return value of func on each successful call.

Raises: TimeoutExpiredError when the elapsed time exceeds wait_timeout.

Iteration Lifecycle

1. A TimeoutWatch is created with timeout=wait_timeout.
2. While remaining time > 0:
3. func(*func_args, **func_kwargs) is called.
4. The return value is yielded to the caller.

5. After the caller processes the yielded value and continues the loop, the sampler sleeps for sleep seconds.

6. If the loop exhausts the timeout without the caller breaking out, TimeoutExpiredError is raised.

Warning: TimeoutSampler does not evaluate the return value of func. The caller must inspect each yielded sample and break or return when a satisfactory value is found. Failing to break out of the loop will always result in TimeoutExpiredError.

Example — Iterate Until Success

from timeout_sampler import TimeoutSampler

def get_pod_status():
    # returns "Pending", "Running", etc.
    ...

for sample in TimeoutSampler(wait_timeout=120, sleep=5, func=get_pod_status):
    if sample == "Running":
        break

Example — Iterate with Logging Disabled

for sample in TimeoutSampler(
    wait_timeout=10,
    sleep=1,
    func=lambda: True,
    print_log=False,
):
    if sample:
        break

Exception Handling During Iteration

When func raises an exception during iteration, TimeoutSampler checks it against exceptions_dict using _should_ignore_exception and _is_exception_matched.

For a detailed walkthrough of the matching algorithm, see How Exception Matching Works.

exceptions_dict Format

{
    ExceptionClass: [message_substring_1, message_substring_2, ...],
    AnotherException: [],   # empty list = match all messages
}

Tip: The match uses isinstance(), so a parent class in exceptions_dict will also catch child classes. See How Exception Matching Works for inheritance examples.

Exception Handling Outcomes

Example — Ignore Specific Exceptions

from timeout_sampler import TimeoutSampler

for sample in TimeoutSampler(
    wait_timeout=30,
    sleep=2,
    func=my_flaky_func,
    exceptions_dict={ConnectionError: [], TimeoutError: []},
):
    if sample:
        break

Example — Filter by Exception Message

from timeout_sampler import TimeoutSampler

for sample in TimeoutSampler(
    wait_timeout=30,
    sleep=2,
    func=my_func,
    exceptions_dict={ValueError: ["not ready", "try again"]},
):
    if sample:
        break

A ValueError("resource not ready") is ignored (contains "not ready"). A ValueError("invalid input") causes an immediate TimeoutExpiredError.

Internal Methods

These methods are not part of the public API but are documented for contributor reference.

_is_exception_matched

@staticmethod
_is_exception_matched(exp: Exception, exception_messages: list[str]) -> bool

Returns: True if exception_messages is empty, or if any non-empty string in the list is a substring of str(exp). False otherwise.

Note: Empty strings in the message list are explicitly skipped — they will not produce a match.

_should_ignore_exception

_should_ignore_exception(self, exp: Exception) -> bool

Returns: True if the exception should be ignored (matches an entry in exceptions_dict via isinstance() and message filtering). False if the exception should be re-raised.

_get_func_info

_get_func_info(self, _func: Callable, type_: str) -> Any

Resolves function metadata (__module__, __name__) for regular, partial, and lambda functions. Used internally to build log messages.

_func_log (property)

@property
_func_log(self) -> str

Returns: A formatted string describing the function call, e.g. "Function: mymodule.my_func Args: (1, 2) Kwargs: {'key': 'val'}". Controlled by print_func_log and print_func_args.

_get_exception_log

_get_exception_log(self, exp: Exception | None = None) -> str

Returns: A multi-line string containing the timeout value, function info (if print_func_log is True), and the last exception class name and message. This string becomes the value attribute of the raised TimeoutExpiredError.

Raised Exceptions

TimeoutSampler raises only one exception type: TimeoutExpiredError.

See TimeoutExpiredError Reference for the full attribute and string-representation reference.

Logging Behavior

Logging is emitted via simple_logger at INFO level.

See Controlling Log Output for usage examples and sample output.

Import Path

from timeout_sampler import TimeoutSampler

The class is exported from the top-level timeout_sampler package (timeout_sampler/__init__.py).

Related Pages

• Polling a Function with TimeoutSampler
• TimeoutExpiredError Reference
• How Exception Matching Works
• Controlling Log Output
• TimeoutWatch API