Prerequisites
- Python 3.9 or later
pip(or any Python package manager)
Install
pip install timeout-sampler
Quick Example
from timeout_sampler import TimeoutSampler
for sample in TimeoutSampler(wait_timeout=10, sleep=2, func=lambda: True):
if sample:
print("Got a truthy value, done!")
break
That's it — TimeoutSampler calls your function every sleep seconds. Each iteration yields the return value so you can inspect it. If wait_timeout seconds elapse without a break, a TimeoutExpiredError is raised.
Step-by-Step Walkthrough
1. Import the essentials
from timeout_sampler import TimeoutSampler, TimeoutExpiredError
2. Define the function you want to poll
Any callable works — a regular function, a lambda, or a method:
import random
def check_service():
"""Simulate a service that becomes ready after a few attempts."""
return random.random() > 0.7
3. Create the sampler and iterate
sampler = TimeoutSampler(
wait_timeout=30, # total seconds to wait
sleep=5, # seconds between retries
func=check_service,
)
for sample in sampler:
if sample:
print("Service is ready!")
break
If check_service() never returns a truthy value within 30 seconds, a TimeoutExpiredError is raised automatically after the loop ends.
4. Handle the timeout
Wrap the loop in a try/except when you need to react to a timeout:
try:
for sample in TimeoutSampler(wait_timeout=10, sleep=2, func=check_service):
if sample:
break
except TimeoutExpiredError as e:
print(f"Timed out: {e}")
TimeoutExpiredError exposes two useful attributes:
| Attribute | Type | Description |
|---|---|---|
last_exp |
Exception | None |
The last exception raised inside func, if any |
elapsed_time |
float | None |
Seconds elapsed before the error was raised |
5. Pass arguments to your function
Use keyword arguments directly on the TimeoutSampler constructor — they are forwarded to func:
def is_ready(host, port):
# ... check connection ...
return True
for sample in TimeoutSampler(
wait_timeout=30,
sleep=5,
func=is_ready,
host="localhost",
port=8080,
):
if sample:
break
Use the @retry Decorator
For the common pattern of "poll until truthy, then return the value," the @retry decorator eliminates the for loop entirely:
from timeout_sampler import retry
@retry(wait_timeout=10, sleep=2)
def get_value():
# return a truthy value when ready
return True
result = get_value() # blocks until truthy or TimeoutExpiredError
If get_value() keeps returning a falsy value for 10 seconds, TimeoutExpiredError is raised. See Retrying Functions with the @retry Decorator for full decorator options.
Advanced Usage
Handling exceptions during polling
By default, TimeoutSampler catches all exceptions raised inside func and keeps retrying. You can control this with exceptions_dict:
for sample in TimeoutSampler(
wait_timeout=20,
sleep=3,
func=check_service,
exceptions_dict={ConnectionError: []},
):
if sample:
break
{ConnectionError: []}— ignore allConnectionErrorinstances (and subclasses) and keep polling.{ConnectionError: ["refused"]}— only ignore aConnectionErrorwhose message contains"refused"; any otherConnectionErrorre-raises immediately.{}— do not ignore any exceptions; every exception re-raises immediately.
Warning: Passing an empty dict
{}means no exceptions are caught. If you want to catch all exceptions (the default), omitexceptions_dictentirely or pass{Exception: []}.
See Filtering and Handling Exceptions for the full inheritance-aware matching rules.
Controlling log output
TimeoutSampler logs elapsed time and function call details by default. Toggle these with three boolean flags:
| Parameter | Default | Effect |
|---|---|---|
print_log |
True |
Log elapsed time on each iteration |
print_func_log |
True |
Include function name and module in log messages |
print_func_args |
True |
Include args/kwargs in the function log |
for sample in TimeoutSampler(
wait_timeout=10,
sleep=2,
func=check_service,
print_log=False, # silence all log output
):
if sample:
break
See Controlling Log Output for details.
Tracking elapsed time independently
The TimeoutWatch helper lets you build custom timing logic outside of TimeoutSampler:
from timeout_sampler import TimeoutWatch
watch = TimeoutWatch(timeout=60.0)
while watch.remaining_time() > 0:
# your custom logic here
pass
See Tracking Elapsed Time with TimeoutWatch for more.
Troubleshooting
| Problem | Cause | Fix |
|---|---|---|
TimeoutExpiredError raised immediately |
wait_timeout is too small or 0 |
Increase wait_timeout to allow at least one poll cycle |
Function arguments not reaching func |
Passing args positionally | Pass arguments as keyword arguments on the TimeoutSampler constructor (e.g., host="localhost") |
| All exceptions are silently swallowed | Default exceptions_dict is {Exception: []} |
Pass a narrower exceptions_dict to only ignore expected exceptions |
| Loop never exits | func returns truthy but there is no break |
Always break (or return) out of the for loop when you get the value you want |
Tip: For copy-paste recipes covering common real-world scenarios, see Common Polling Patterns.