Spaces:
Sleeping
Sleeping
File size: 1,609 Bytes
a42ab7e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 |
"""
Rate Limiting Dependencies
Functions for checking and enforcing rate limits on API endpoints.
"""
import logging
from datetime import datetime, timedelta
from sqlalchemy import select, and_
from sqlalchemy.ext.asyncio import AsyncSession
from core.models import RateLimit
logger = logging.getLogger(__name__)
async def check_rate_limit(
db: AsyncSession,
identifier: str,
endpoint: str,
limit: int,
window_minutes: int
) -> bool:
"""
Check if request is within rate limits.
Returns True if allowed, False if limit exceeded.
"""
now = datetime.utcnow()
window_start = now - timedelta(minutes=window_minutes)
# Check existing limit (get most recent if multiple exist)
query = select(RateLimit).where(
and_(
RateLimit.identifier == identifier,
RateLimit.endpoint == endpoint,
RateLimit.window_start >= window_start
)
).order_by(RateLimit.window_start.desc())
result = await db.execute(query)
rate_limit = result.scalars().first()
if rate_limit:
if rate_limit.attempts >= limit:
return False
# Increment attempts
rate_limit.attempts += 1
await db.commit()
return True
else:
# Create new rate limit record
new_limit = RateLimit(
identifier=identifier,
endpoint=endpoint,
attempts=1,
window_start=now,
expires_at=now + timedelta(minutes=window_minutes)
)
db.add(new_limit)
await db.commit()
return True
|