Spaces:
Paused
Paused
| 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 --- | |
| 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 | |
| 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 | |
| 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 | |