(
user_id: str, service: str,
is_pro: bool = False, is_max: bool = False, is_plus: bool = False
)
| 148 | return int(val) >= RATE_LIMIT_PER_MINUTE if val else False |
| 149 | |
| 150 | async def check_usage( |
| 151 | user_id: str, service: str, |
| 152 | is_pro: bool = False, is_max: bool = False, is_plus: bool = False |
| 153 | ) -> bool: |
| 154 | if await is_rate_limited(user_id): |
| 155 | asyncio.create_task(_send_abuse_alert_async(user_id, service)) |
| 156 | return True |
| 157 | |
| 158 | r = await get_redis() |
| 159 | val = await r.get(f"quota:{user_id}:{service}") |
| 160 | current_usage = int(val) if val else 0 |
| 161 | |
| 162 | if is_max: |
| 163 | limit = MAX_QUOTA_LIMITS[service] |
| 164 | elif is_pro: |
| 165 | limit = PRO_QUOTA_LIMITS[service] |
| 166 | elif is_plus: |
| 167 | limit = PLUS_QUOTA_LIMITS[service] |
| 168 | else: |
| 169 | limit = QUOTA_LIMITS[service] |
| 170 | |
| 171 | return current_usage >= limit |
| 172 | |
| 173 | async def get_all_usage_data() -> dict: |
| 174 | r = await get_redis() |
no test coverage detected