Rate limiting that approaches but never exceeds a threshold. Classic token-bucket limiter with optional hold_at_threshold() mode for steady approach without burst overshoot.
pip install edgingfrom edging import TokenBucket, RateLimitExceeded
bucket = TokenBucket(rate=5.0, capacity=5.0).hold_at_threshold(True)
assert bucket.acquire(1.0) is True
assert bucket.tokens() < 5.0
hit = None
for _ in range(20):
try:
bucket.acquire(1.0, blocking=False)
except RateLimitExceeded as e:
hit = e
break
assert hit is not None and hit.retry_after > 0- Cap throughput without silent overshoot past capacity
- Non-blocking acquire raises
RateLimitExceededwithretry_after - Thread-safe token accounting
| Symbol | Description |
|---|---|
TokenBucket(rate, capacity) |
Token-bucket limiter |
hold_at_threshold(enabled) |
Steady approach mode; chainable |
acquire(...) |
Take tokens; blocking or raise |
RateLimitExceeded |
Non-blocking failure with .retry_after |
Part of the coldbricks micro-library suite (small, typed, zero-dependency helpers):
| Package | Role |
|---|---|
buzzkill |
Guaranteed shutdown callbacks |
aftercare |
Reverse-order teardown |
fluffer |
Startup pre-warm |
| edging | Token-bucket rate limit |
readback |
Echo-confirmation validation |
hearback |
Output re-validation decorator |
agecheck |
Timezone-correct age thresholds |
transponder |
State broadcast on change |
pip install -e ".[dev]"
pytest -q
ruff check .
mypy srcThis is an early 0.1.0 release (Alpha) with active development planned:
- Async acquire APIs
- Shared / distributed backends
- Adaptive threshold policies
MIT © Coldbricks — coldbricks@gmail.com