api_keys.py 1.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556
  1. """API Key 管理端点。
  2. 路由前缀: /api/v1/api-keys
  3. """
  4. from fastapi import APIRouter, Depends, HTTPException
  5. from pydantic import BaseModel
  6. from app.core.auth import get_current_user
  7. from app.services import api_key_service
  8. router = APIRouter()
  9. class CreateApiKeyRequest(BaseModel):
  10. name: str = "default"
  11. class ApiKeyResponse(BaseModel):
  12. id: str
  13. key: str # 创建时返回完整 key,列表时返回 masked
  14. name: str
  15. status: str | None = None
  16. last_used_at: str | None = None
  17. created_at: str | None = None
  18. @router.post("/", response_model=ApiKeyResponse)
  19. async def create_key(
  20. req: CreateApiKeyRequest,
  21. current_user: dict = Depends(get_current_user),
  22. ):
  23. """创建新的 API Key(完整 key 仅返回一次)。"""
  24. user_id = current_user.get("sub")
  25. result = await api_key_service.create_api_key(user_id, req.name)
  26. return ApiKeyResponse(**result)
  27. @router.get("/", response_model=list[ApiKeyResponse])
  28. async def list_keys(current_user: dict = Depends(get_current_user)):
  29. """列出当前用户的所有 API Key。"""
  30. user_id = current_user.get("sub")
  31. keys = await api_key_service.list_api_keys(user_id)
  32. return [ApiKeyResponse(**k) for k in keys]
  33. @router.delete("/{key_id}")
  34. async def revoke_key(
  35. key_id: str,
  36. current_user: dict = Depends(get_current_user),
  37. ):
  38. """吊销指定的 API Key。"""
  39. user_id = current_user.get("sub")
  40. result = await api_key_service.revoke_api_key(key_id, user_id)
  41. if "error" in result:
  42. raise HTTPException(status_code=404, detail=result["error"])
  43. return result