返回 (allowed, remaining, retry_after_seconds)。
(self, key, limit, window)
| 19 | self._lock = threading.Lock() |
| 20 | |
| 21 | def check(self, key, limit, window): |
| 22 | """返回 (allowed, remaining, retry_after_seconds)。""" |
| 23 | now = time.time() |
| 24 | cutoff = now - window |
| 25 | with self._lock: |
| 26 | q = self._hits.setdefault(key, []) |
| 27 | while q and q[0] < cutoff: |
| 28 | q.pop(0) |
| 29 | if len(q) >= limit: |
| 30 | retry = window - (now - q[0]) |
| 31 | return False, 0, max(0.0, retry) |
| 32 | q.append(now) |
| 33 | return True, limit - len(q), 0.0 |
| 34 | |
| 35 | |
| 36 | _limiter = SlidingWindowLimiter() |
no test coverage detected