INNER CODE UNIT · Python
actualised_previous_ttl
alisaifee/limits · limits/aio/storage/memcached/__init__.py:150
actualised_previous_ttl = max(0, previous_ttl - (t1 - t0))
weighted_count = (
previous_count * actualised_previous_ttl / expiry + current_count
)
if floor(weighted_count) > limit:
# Another hit won the race condition: revert the increment and refuse this hit
# Limitation: during high concurrency at the end of the window,
# the counter is shifted and cannot be decremented, so less requests than expected are allowed.
await self.bridge.decr(current_key, amount, noreply=True)
return False
return True
async def get_sliding_window(
self, key: str, expiry: int
) -> tuple[int, float, int, float]:
now = time.time()
previous_key, current_key = self.sliding_window_keys(key, expiry, now)
return await self._get_sliding_window_info(