zenith-backend / app /routers /api_keys.py
teoat's picture
fix(backend): fix port and health check robustness
d29a5a0 verified
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