INNER CODE UNIT · Python
clear_sliding_window
alisaifee/limits · limits/aio/storage/memcached/__init__.py:171
async def clear_sliding_window(self, key: str, expiry: int) -> None:
now = time.time()
previous_key, current_key = self.sliding_window_keys(key, expiry, now)
await asyncio.gather(self.clear(previous_key), self.clear(current_key))
async def _get_sliding_window_info(
self, previous_key: str, current_key: str, expiry: int, now: float
) -> tuple[int, float, int, float]:
result = await self.bridge.get_many([previous_key, current_key])
previous_count = result.get(previous_key.encode("utf-8"), 0)
current_count = result.get(current_key.encode("utf-8"), 0)
if previous_count == 0:
previous_ttl = float(0)
else:
previous_ttl = (1 - (((now - expiry) / expiry) % 1)) * expiry
current_ttl = (1 - ((now / expiry) % 1)) * expiry + expiry