Spaces:
Paused
Paused
File size: 6,853 Bytes
4efde5d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 | """
API Keys API Endpoints
This module provides REST API endpoints for managing API keys:
- POST /api/api-keys - Create a new API key
- GET /api/api-keys - List all API keys for the authenticated user
- DELETE /api/api-keys/{key_id} - Revoke/delete an API key
"""
from fastapi import APIRouter, Depends, HTTPException
from typing import List
from uuid import UUID
from services.api_keys import (
APIKeyService,
APIKeyCreateRequest,
APIKeyResponse,
APIKeyCreateResponse,
)
from services.supabase import DBConnection
from utils.auth_utils import get_current_user_id_from_jwt
from utils.logger import logger
router = APIRouter()
async def get_api_key_service() -> APIKeyService:
"""Dependency to get API key service instance"""
db = DBConnection()
await db.initialize()
return APIKeyService(db)
async def get_account_id_from_user_id(user_id: str) -> UUID:
"""Get account ID from user ID using basejump accounts table"""
try:
db = DBConnection()
await db.initialize()
client = await db.client
# Query the basejump.accounts table for the user's primary account
result = (
await client.schema("basejump")
.table("accounts")
.select("id")
.eq("primary_owner_user_id", user_id)
.eq("personal_account", True) # Get the user's personal account
.limit(1)
.execute()
)
if not result.data:
raise HTTPException(status_code=404, detail="User account not found")
return UUID(result.data[0]["id"])
except Exception as e:
logger.error(f"Error getting account ID: {e}")
raise HTTPException(status_code=500, detail="Failed to get user account")
@router.post("/api-keys", response_model=APIKeyCreateResponse)
async def create_api_key(
request: APIKeyCreateRequest,
user_id: str = Depends(get_current_user_id_from_jwt),
api_key_service: APIKeyService = Depends(get_api_key_service),
):
"""
Create a new API key for the authenticated user
Args:
request: API key creation request with title, description, and expiration
user_id: Authenticated user ID from JWT or API key
api_key_service: API key service instance
Returns:
APIKeyCreateResponse: The newly created API key details including the key value
"""
try:
account_id = await get_account_id_from_user_id(user_id)
logger.debug(
"Creating API key",
user_id=user_id,
account_id=str(account_id),
title=request.title,
)
api_key = await api_key_service.create_api_key(account_id, request)
logger.debug(
"API key created successfully",
user_id=user_id,
key_id=str(api_key.key_id),
title=api_key.title,
)
return api_key
except HTTPException:
raise
except Exception as e:
logger.error(f"Unexpected error creating API key: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Failed to create API key")
@router.get("/api-keys", response_model=List[APIKeyResponse])
async def list_api_keys(
user_id: str = Depends(get_current_user_id_from_jwt),
api_key_service: APIKeyService = Depends(get_api_key_service),
):
"""
List all API keys for the authenticated user
Args:
user_id: Authenticated user ID from JWT or API key
api_key_service: API key service instance
Returns:
List[APIKeyResponse]: List of API keys (without the actual key values)
"""
try:
account_id = await get_account_id_from_user_id(user_id)
logger.debug("Listing API keys", user_id=user_id, account_id=str(account_id))
api_keys = await api_key_service.list_api_keys(account_id)
logger.debug(
"API keys listed successfully", user_id=user_id, count=len(api_keys)
)
return api_keys
except HTTPException:
raise
except Exception as e:
logger.error(f"Unexpected error listing API keys: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Failed to list API keys")
@router.patch("/api-keys/{key_id}/revoke")
async def revoke_api_key(
key_id: UUID,
user_id: str = Depends(get_current_user_id_from_jwt),
api_key_service: APIKeyService = Depends(get_api_key_service),
):
"""
Revoke an API key
Args:
key_id: The ID of the API key to revoke
user_id: Authenticated user ID from JWT or API key
api_key_service: API key service instance
Returns:
dict: Success message
"""
try:
account_id = await get_account_id_from_user_id(user_id)
logger.debug(
"Revoking API key",
user_id=user_id,
account_id=str(account_id),
key_id=str(key_id),
)
success = await api_key_service.revoke_api_key(account_id, key_id)
if success:
logger.debug(
"API key revoked successfully", user_id=user_id, key_id=str(key_id)
)
return {"message": "API key revoked successfully"}
else:
raise HTTPException(status_code=500, detail="Failed to revoke API key")
except HTTPException:
raise
except Exception as e:
logger.error(f"Unexpected error revoking API key: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Failed to revoke API key")
@router.delete("/api-keys/{key_id}")
async def delete_api_key(
key_id: UUID,
user_id: str = Depends(get_current_user_id_from_jwt),
api_key_service: APIKeyService = Depends(get_api_key_service),
):
"""
Permanently delete an API key
Args:
key_id: The ID of the API key to delete
user_id: Authenticated user ID from JWT or API key
api_key_service: API key service instance
Returns:
dict: Success message
"""
try:
account_id = await get_account_id_from_user_id(user_id)
logger.debug(
"Deleting API key",
user_id=user_id,
account_id=str(account_id),
key_id=str(key_id),
)
success = await api_key_service.delete_api_key(account_id, key_id)
if success:
logger.debug(
"API key deleted successfully", user_id=user_id, key_id=str(key_id)
)
return {"message": "API key deleted successfully"}
else:
raise HTTPException(status_code=500, detail="Failed to delete API key")
except HTTPException:
raise
except Exception as e:
logger.error(f"Unexpected error deleting API key: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Failed to delete API key")
|