from datetime import datetime from typing import Optional from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel, Field from sqlalchemy.orm import Session from app.services.infrastructure.api_key_service import api_key_service from app.services.infrastructure.auth_service import auth_service from core.database import User, get_db router = APIRouter() # --- Schemas --- class APIKeyCreate(BaseModel): name: str = Field(..., min_length=1, max_length=100) description: Optional[str] = Field(None, max_length=255) expires_days: int = Field(default=365, ge=1, le=3650) permissions: list[str] = Field(default=["read"]) class APIKeyResponse(BaseModel): id: str prefix: str name: str description: Optional[str] user_id: str permissions: list[str] is_active: bool created_at: datetime expires_at: Optional[datetime] last_used_at: Optional[datetime] class Config: from_attributes = True class APIKeyCreatedResponse(APIKeyResponse): api_key: str # Only shown once on creation # --- Endpoints --- @router.post("/", response_model=APIKeyCreatedResponse, status_code=status.HTTP_201_CREATED) async def create_api_key( key_data: APIKeyCreate, current_user: User = Depends(auth_service.get_current_user), db: Session = Depends(get_db) ): """Create a new API key for the current user""" raw_key, api_key_model = api_key_service.generate_api_key( db=db, user_id=current_user.id, name=key_data.name, description=key_data.description, permissions=key_data.permissions, expires_days=key_data.expires_days ) response = APIKeyCreatedResponse.from_orm(api_key_model) response.api_key = raw_key return response @router.get("/", response_model=list[APIKeyResponse]) async def list_api_keys( current_user: User = Depends(auth_service.get_current_user), db: Session = Depends(get_db) ): """List all active API keys for the current user""" keys = api_key_service.list_keys(db, current_user.id) return keys @router.delete("/{key_id}", status_code=status.HTTP_204_NO_CONTENT) async def revoke_api_key( key_id: str, current_user: User = Depends(auth_service.get_current_user), db: Session = Depends(get_db) ): """Revoke (deactivate) an API key""" success = api_key_service.revoke_key(db, key_id, current_user.id) if not success: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="API key not found or already revoked" ) return None