Spaces:
Paused
Paused
File size: 2,667 Bytes
5ec8557 | 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 | import uuid
from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Session
from auth import generate_api_key, require_auth
from db import get_db
from models import APIKey, User
from schemas import MessageResponse
router = APIRouter(prefix="/api-keys", tags=["api-keys"])
class CreateKeyRequest(BaseModel):
label: str
class CreateKeyResponse(BaseModel):
id: str
key: str
label: str
key_prefix: str
created_at: datetime
class KeyListItem(BaseModel):
id: str
label: str
key_prefix: str
created_at: datetime
last_used_at: datetime | None
model_config = {"from_attributes": True}
@router.get("", response_model=list[KeyListItem])
def list_keys(user: User = Depends(require_auth), db: Session = Depends(get_db)):
keys = (
db.execute(
select(APIKey)
.where(APIKey.created_by == user.id, APIKey.revoked_at.is_(None))
.order_by(APIKey.created_at.desc())
)
.scalars()
.all()
)
return [
KeyListItem(
id=str(k.id),
label=k.label,
key_prefix=k.key_prefix,
created_at=k.created_at,
last_used_at=k.last_used_at,
)
for k in keys
]
@router.post("", response_model=CreateKeyResponse, status_code=201)
def create_key(body: CreateKeyRequest, user: User = Depends(require_auth), db: Session = Depends(get_db)):
full_key, prefix, key_hash = generate_api_key()
api_key = APIKey(key_prefix=prefix, key_hash=key_hash, label=body.label, created_by=user.id)
db.add(api_key)
db.commit()
db.refresh(api_key)
return CreateKeyResponse(
id=str(api_key.id),
key=full_key,
label=api_key.label,
key_prefix=prefix,
created_at=api_key.created_at,
)
@router.delete("/{key_id}", response_model=MessageResponse)
def revoke_key(key_id: str, user: User = Depends(require_auth), db: Session = Depends(get_db)):
try:
key_uuid = uuid.UUID(key_id)
except (TypeError, ValueError):
raise HTTPException(status_code=404, detail="API key not found.")
api_key = db.get(APIKey, key_uuid)
if api_key is None or api_key.created_by != user.id:
raise HTTPException(status_code=404, detail="API key not found.")
if api_key.revoked_at is not None:
raise HTTPException(status_code=400, detail="API key is already revoked.")
api_key.revoked_at = datetime.now(timezone.utc)
db.commit()
return MessageResponse(message="API key revoked.") |