quivr/backend/auth/api_key_handler.py
Zineb El Bachiri 9aaedcff51
Fix/requests limiting (#980)
* 🗃️ Rename users table into user_daily_usage

* 💥 replace User model with UserIdentity model

* 🗃️ New UserDailyUsage class for database interaction

* 🐛 fix daily requests rate limiting per user

* 🐛 fix user stats and properties update

* ✏️ add typing and linting

* 🚚 rename user_dialy_usage Class  into user_usage & requests_count into daily_requests_count

* 🚑 fix some rebase errors
2023-08-21 14:05:13 +02:00

51 lines
1.5 KiB
Python

from datetime import datetime
from uuid import UUID
from fastapi import HTTPException
from models.settings import get_supabase_db
from models.user_identity import UserIdentity
from pydantic import DateError
async def verify_api_key(
api_key: str,
) -> bool:
try:
# Use UTC time to avoid timezone issues
current_date = datetime.utcnow().date()
supabase_db = get_supabase_db()
result = supabase_db.get_active_api_key(UUID(api_key))
if result.data is not None and len(result.data) > 0:
api_key_creation_date = datetime.strptime(
result.data[0]["creation_time"], "%Y-%m-%dT%H:%M:%S"
).date()
# Check if the API key was created in the month of the current date
if (api_key_creation_date.month == current_date.month) and (
api_key_creation_date.year == current_date.year
):
return True
return False
except DateError:
return False
async def get_user_from_api_key(
api_key: str,
) -> UserIdentity:
supabase_db = get_supabase_db()
# Lookup the user_id from the api_keys table
user_id_data = supabase_db.get_user_id_by_api_key(UUID(api_key))
if not user_id_data.data:
raise HTTPException(status_code=400, detail="Invalid API key.")
user_id = user_id_data.data[0]["user_id"]
# Lookup the email from the users table. Todo: remove and use user_id for credentials
email = supabase_db.get_user_email(user_id)
return UserIdentity(email=email, id=user_id)