Spaces:

Jindrich3
/

openajaj

Running

App Files Files Community

openajaj / rate_limit.py

Jindrich3

hynek med

Super-squash branch 'main' using huggingface_hub

b356196 1 day ago

Raw History Blame Contribute Delete

2.51 kB

"""Per-client request caps, in memory.
Keys are exact paths, not prefixes. `"/api/stt"` used to match by prefix, which
swallowed `/api/stt/check` and `/api/stt/usage` as well: a single /transcribe
load spends three of those, so the fourth reload inside a minute was refused
and the page reported the transcription service as down. The websocket it was
written for never reaches this code at all — `BaseHTTPMiddleware` skips
non-http scopes, so websocket limits live in `stt_budget` instead.
"""

import threading
import time

# exact path → (max_requests, window_seconds)
LIMITS: dict[str, tuple[int, int]] = {
    "/api/chat": (15, 60),
    # Voice mode sends one request per sentence, prefetched as the answer
    # streams, so a low cap counts sentences rather than users: at 10 the
    # eleventh sentence of one answer was refused and the reply finished in the
    # browser's built-in voice. edge-tts is unmetered and the result is cached,
    # so this only exists to stop the Space being used as a free TTS proxy.
    "/api/tts": (60, 60),
    "/api/benchmark": (1, 300),
    # Every call makes an outbound request to Deepgram's auth endpoint, so it
    # cannot go uncapped. The chat page calls it once per load, /t after each
    # unexpected socket close: reconnect backoff (1 s doubling to 30 s) keeps
    # that to about seven in the first minute, then two a minute.
    "/api/stt/check": (30, 60),
    "/api/stt/usage": (30, 60),
    # /t asks once per socket, like the check; each call mints a Deepgram token
    "/api/stt/token": (30, 60),
    # /t reports its visit every 5 min and on hide: far under this
    "/api/stt/session": (30, 60),
    "/api/captcha/challenge": (10, 60),
    "/api/captcha/verify": (10, 60),
}

_buckets: dict[str, list[float]] = {}
_lock = threading.Lock()


def is_limited(client: str, path: str, now: float | None = None) -> bool:
    """True when this client has used up `path`'s allowance. Counts the call."""
    limit_window = LIMITS.get(path)
    if limit_window is None:
        return False
    limit, window = limit_window
    now = time.time() if now is None else now
    key = f"{client}:{path}"
    with _lock:
        recent = [t for t in _buckets.get(key, ()) if t > now - window]
        if len(recent) >= limit:
            _buckets[key] = recent
            return True
        recent.append(now)
        _buckets[key] = recent
    return False


def reset() -> None:
    """Drop all buckets. For tests."""
    with _lock:
        _buckets.clear()

	"""Per-client request caps, in memory.

	Keys are exact paths, not prefixes. `"/api/stt"` used to match by prefix, which
	swallowed `/api/stt/check` and `/api/stt/usage` as well: a single /transcribe
	load spends three of those, so the fourth reload inside a minute was refused
	and the page reported the transcription service as down. The websocket it was
	written for never reaches this code at all — `BaseHTTPMiddleware` skips
	non-http scopes, so websocket limits live in `stt_budget` instead.
	"""

	import threading
	import time

	# exact path → (max_requests, window_seconds)
	LIMITS: dict[str, tuple[int, int]] = {
	"/api/chat": (15, 60),
	# Voice mode sends one request per sentence, prefetched as the answer
	# streams, so a low cap counts sentences rather than users: at 10 the
	# eleventh sentence of one answer was refused and the reply finished in the
	# browser's built-in voice. edge-tts is unmetered and the result is cached,
	# so this only exists to stop the Space being used as a free TTS proxy.
	"/api/tts": (60, 60),
	"/api/benchmark": (1, 300),
	# Every call makes an outbound request to Deepgram's auth endpoint, so it
	# cannot go uncapped. The chat page calls it once per load, /t after each
	# unexpected socket close: reconnect backoff (1 s doubling to 30 s) keeps
	# that to about seven in the first minute, then two a minute.
	"/api/stt/check": (30, 60),
	"/api/stt/usage": (30, 60),
	# /t asks once per socket, like the check; each call mints a Deepgram token
	"/api/stt/token": (30, 60),
	# /t reports its visit every 5 min and on hide: far under this
	"/api/stt/session": (30, 60),
	"/api/captcha/challenge": (10, 60),
	"/api/captcha/verify": (10, 60),
	}

	_buckets: dict[str, list[float]] = {}
	_lock = threading.Lock()


	def is_limited(client: str, path: str, now: float \| None = None) -> bool:
	"""True when this client has used up `path`'s allowance. Counts the call."""
	limit_window = LIMITS.get(path)
	if limit_window is None:
	return False
	limit, window = limit_window
	now = time.time() if now is None else now
	key = f"{client}:{path}"
	with _lock:
	recent = [t for t in _buckets.get(key, ()) if t > now - window]
	if len(recent) >= limit:
	_buckets[key] = recent
	return True
	recent.append(now)
	_buckets[key] = recent
	return False


	def reset() -> None:
	"""Drop all buckets. For tests."""
	with _lock:
	_buckets.clear()