MediumRate LimitingPython 3
Notification Rate Limiter
Throttle noisy notification channels while deduping repeated webhook or automation events.
35m2 sample tests5 hidden tests
Implement NotificationLimiter, a per-channel notification throttle with dedupe keys.
Requirements
- Constructor receives
limit,window_seconds, and injectednow. allow(channel, dedupe_key)returns(allowed, retry_after_seconds).- A channel may send at most
limitunique dedupe keys per rolling window. - Repeating the same dedupe key inside the window is blocked without consuming a new slot.
- Expired records are removed on every call.
retry_after_secondsis0when allowed.
Example
python
1clock = {"now": 0.0}
2limiter = NotificationLimiter(2, 10.0, lambda: clock["now"])
3assert limiter.allow("slack", "a") == (True, 0.0)
4assert limiter.allow("slack", "a")[0] is FalseConstraints
- Use injected time. Keep tests free of sleeps.
- Keep channels isolated.
Editor