43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
"""Simple in-memory rate limiter per user."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import defaultdict
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
|
|
from app.api.deps import require_student
|
|
from app.core.schemas import UserProfile
|
|
|
|
|
|
class RateLimiter:
|
|
def __init__(self, max_requests: int, window_seconds: int) -> None:
|
|
self._max = max_requests
|
|
self._window = window_seconds
|
|
self._buckets: dict[int, list[float]] = defaultdict(list)
|
|
|
|
def check(self, user_id: int) -> None:
|
|
now = time.time()
|
|
cutoff = now - self._window
|
|
bucket = self._buckets[user_id]
|
|
# purge expired entries
|
|
while bucket and bucket[0] < cutoff:
|
|
bucket.pop(0)
|
|
if len(bucket) >= self._max:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail=f"请求过于频繁，请 {self._window // 60} 分钟后重试",
|
|
)
|
|
bucket.append(now)
|
|
|
|
|
|
_diagnosis_limiter = RateLimiter(max_requests=5, window_seconds=60)
|
|
|
|
|
|
def check_diagnosis_rate_limit(
|
|
user: Annotated[UserProfile, Depends(require_student)],
|
|
) -> None:
|
|
_diagnosis_limiter.check(user.id)
|