This package provides a simple throttle controller for general use-cases. For example, you can use this package to throttle API requests to avoid rate-limiting.
from throttle_controller import SimpleThrottleController throttle = SimpleThrottleController.create(default_cooldown_time=3.0) throttle.wait_if_needed("http://example.com/path/to/api") throttle.record_use_time_as_now("http://example.com/path/to/api") ... # requests throttle.wait_if_needed("http://example.com/path/to/api") # wait 3.0 seconds throttle.record_use_time_as_now("http://example.com/path/to/api")
from throttle_controller import SimpleThrottleController throttle = SimpleThrottleController.create(default_cooldown_time=3.0) for _ in range(10): with throttle.use("http://example.com/path/to/api"): # wait if cooldown needed requests.get("http://example.com/path/to/api")
Pass now when you need deterministic tests or replayable behavior:
import datetime from throttle_controller import SimpleThrottleController fixed_time = datetime.datetime(2026, 1, 2, 3, 4, 5) throttle = SimpleThrottleController.create( default_cooldown_time=3.0, now=lambda: fixed_time, )
This package is currently in Alpha (Development Status :: 3 - Alpha).
- Breaking changes may occur between minor versions while in Alpha.
- The package will be considered for Beta promotion when thread-safe and multi-process support is implemented.
- A stable (1.0) release is planned after Beta validation.
The use() context manager and the manual record_use_time_as_now() call record
the start of each use, not the end. This means the cooldown is measured
start-to-start: the next call may proceed once cooldown_time has elapsed
since the start of the previous call.
call 1 start ──────────── call 1 body (2.9 s) ────── call 1 end
│
└── cooldown_time = 3.0 s ──────────────────────────► call 2 may start
(only 0.1 s after call 1 ends)
If you need end-to-start behaviour (next call waits until at least
cooldown_time after the previous call finishes), record the time after your
work completes instead of relying on the context manager:
throttle.wait_if_needed(key) result = do_work() # your code here throttle.record_use_time_as_now(key) # record end of work
Currently this package supports only to use in single thread / single process use-cases.
Reusing the same controller instance from multiple threads raises RuntimeError.
This repository uses lefthook to run the same checks as CI locally, so problems surface before they reach CI.
# Install dependencies uv sync # Install the Git hooks (once; requires lefthook on your PATH) lefthook install
Once installed, the hooks run automatically:
- pre-commit:
uv run poe check - pre-push:
uv run poe checkanduv run poe test
You can also run the checks manually:
uv run poe check
uv run poe testCI still runs the full matrix (see .github/workflows/); the hooks only bring that
feedback earlier on your machine.
The 3-Clause BSD License. See also LICENSE file.