app/services/auth_service.py

import hashlib
import secrets
import string
import bcrypt
from datetime import datetime, timezone
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models.db_models import APIKey


def generate_api_key() -> tuple[str, str]:
    alphabet = string.ascii_letters + string.digits
    raw = "sk-" + "".join(secrets.choice(alphabet) for _ in range(40))
    prefix = raw[:8]
    return raw, prefix


def hash_key(raw_key: str) -> str:
    return hashlib.sha256(raw_key.encode()).hexdigest()


def verify_key(raw_key: str, stored_hash: str) -> bool:
    return hash_key(raw_key) == stored_hash


async def get_api_key(db: AsyncSession, raw_key: str) -> APIKey | None:
    key_hash = hash_key(raw_key)
    result = await db.execute(
        select(APIKey).where(APIKey.key_hash == key_hash, APIKey.is_active == True)
    )
    return result.scalar_one_or_none()


async def validate_api_key(db: AsyncSession, raw_key: str) -> APIKey | None:
    api_key = await get_api_key(db, raw_key)
    if api_key is None:
        return None
    api_key.last_used_at = datetime.now(timezone.utc)
    api_key.total_requests += 1
    await db.flush()
    return api_key