"""API Key 管理:生成 / 列表 / 吊销。""" import datetime as dt from typing import List from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from ..database import get_db from ..deps import get_current_user from ..models import User, ApiKey from ..schemas import ApiKeyCreateIn, ApiKeyCreatedOut, ApiKeyOut from ..security import generate_api_key router = APIRouter(prefix="/api/keys", tags=["api-keys"]) @router.post("", response_model=ApiKeyCreatedOut) def create_key( body: ApiKeyCreateIn, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): """生成一个新的 API Key。明文 key 仅此一次返回,库内存哈希。""" raw = generate_api_key() api_key = ApiKey( user_id=current_user.id, key_hash=ApiKey.hash_key(raw), key_prefix=raw[:10], name=body.name, ) db.add(api_key) db.commit() db.refresh(api_key) data = ApiKeyOut.model_validate(api_key).model_dump() return ApiKeyCreatedOut(**data, key=raw) @router.get("", response_model=List[ApiKeyOut]) def list_keys( current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): keys = ( db.query(ApiKey) .filter(ApiKey.user_id == current_user.id) .order_by(ApiKey.created_at.desc()) .all() ) return keys @router.post("/{key_id}/revoke", response_model=ApiKeyOut) def revoke_key( key_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): api_key = ( db.query(ApiKey) .filter(ApiKey.id == key_id, ApiKey.user_id == current_user.id) .first() ) if api_key is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "API Key 不存在") api_key.revoked = True db.commit() db.refresh(api_key) return api_key