bstein-dev-home/backend/atlas_portal/rate_limit.py

from __future__ import annotations

import time

from . import settings

_RATE_BUCKETS: dict[str, dict[str, list[float]]] = {}


def rate_limit_allow(ip: str, *, key: str, limit: int, window_sec: int) -> bool:
    if limit <= 0:
        return True
    now = time.time()
    window_start = now - window_sec
    buckets_by_ip = _RATE_BUCKETS.setdefault(key, {})
    bucket = buckets_by_ip.setdefault(ip, [])
    bucket[:] = [t for t in bucket if t >= window_start]
    if len(bucket) >= limit:
        return False
    bucket.append(now)
    return True