Samuel AmarandClaude Haiku 4.5
77d7a50fa9
- FastAPI async gateway with httpx proxying to multiple upstreams - SQLite database with SQLAlchemy ORM - Admin console: manage services, users, API keys, endpoint access - Per-key, per-endpoint granular access control - OpenAPI document sync and caching (5-minute TTL) - Request/response logging with full transaction inspection - In-memory rate limiting (per-key, fixed-window) - Tiered log retention (7d payloads, 90d rows, incremental vacuum) - TLS verification toggle per service (for self-signed certificates) - Service connectivity validation with automatic endpoint refresh - Request browser with filters and deep-link inspection - Docker setup with persistent volume - Modal forms for create/edit flows Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
27 lines
799 B
Python
27 lines
799 B
Python
"""In-memory fixed-window rate limiter, keyed per API key.
|
|
|
|
Good enough for a single-process gateway; swap for Redis if you scale out.
|
|
"""
|
|
import time
|
|
from collections import defaultdict
|
|
from threading import Lock
|
|
|
|
_windows: dict[int, tuple[int, int]] = defaultdict(lambda: (0, 0)) # key_id -> (window_start, count)
|
|
_lock = Lock()
|
|
|
|
|
|
def check(key_id: int, limit_per_minute: int) -> bool:
|
|
"""Returns True if the request is allowed."""
|
|
if limit_per_minute <= 0:
|
|
return True
|
|
now = int(time.time() // 60)
|
|
with _lock:
|
|
window, count = _windows[key_id]
|
|
if window != now:
|
|
_windows[key_id] = (now, 1)
|
|
return True
|
|
if count >= limit_per_minute:
|
|
return False
|
|
_windows[key_id] = (window, count + 1)
|
|
return True
|