default
d54e53e0b7
FastAPI + SQLAlchemy + asyncio TCP 服务器，支持设备管理、实时定位、 告警、考勤打卡、蓝牙记录、指令下发、TTS语音播报等功能。
143 lines
4.4 KiB
Python
143 lines
4.4 KiB
Python
"""
|
||
API Keys Router - API密钥管理接口
|
||
Endpoints for creating, listing, updating, and deactivating API keys.
|
||
Admin permission required.
|
||
"""
|
||
|
||
import secrets
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.database import get_db
|
||
from app.dependencies import require_admin, _hash_key
|
||
from app.models import ApiKey
|
||
from app.schemas import (
|
||
APIResponse,
|
||
ApiKeyCreate,
|
||
ApiKeyCreateResponse,
|
||
ApiKeyResponse,
|
||
ApiKeyUpdate,
|
||
PaginatedList,
|
||
)
|
||
|
||
import math
|
||
|
||
router = APIRouter(prefix="/api/keys", tags=["API Keys / 密钥管理"])
|
||
|
||
|
||
def _generate_key() -> str:
|
||
"""Generate a random 32-char hex API key."""
|
||
return secrets.token_hex(16)
|
||
|
||
|
||
@router.get(
|
||
"",
|
||
response_model=APIResponse[PaginatedList[ApiKeyResponse]],
|
||
summary="列出API密钥 / List API keys",
|
||
dependencies=[Depends(require_admin)],
|
||
)
|
||
async def list_keys(
|
||
page: int = Query(default=1, ge=1),
|
||
page_size: int = Query(default=20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""列出所有API密钥（不返回密钥值）/ List all API keys (key values are never shown)."""
|
||
count_result = await db.execute(select(func.count(ApiKey.id)))
|
||
total = count_result.scalar() or 0
|
||
|
||
offset = (page - 1) * page_size
|
||
result = await db.execute(
|
||
select(ApiKey).order_by(ApiKey.created_at.desc()).offset(offset).limit(page_size)
|
||
)
|
||
keys = list(result.scalars().all())
|
||
|
||
return APIResponse(
|
||
data=PaginatedList(
|
||
items=[ApiKeyResponse.model_validate(k) for k in keys],
|
||
total=total,
|
||
page=page,
|
||
page_size=page_size,
|
||
total_pages=math.ceil(total / page_size) if total else 0,
|
||
)
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"",
|
||
response_model=APIResponse[ApiKeyCreateResponse],
|
||
status_code=201,
|
||
summary="创建API密钥 / Create API key",
|
||
dependencies=[Depends(require_admin)],
|
||
)
|
||
async def create_key(body: ApiKeyCreate, db: AsyncSession = Depends(get_db)):
|
||
"""创建新的API密钥。明文密钥仅在创建时返回一次。
|
||
Create a new API key. The plaintext key is returned only once."""
|
||
raw_key = _generate_key()
|
||
key_hash = _hash_key(raw_key)
|
||
|
||
db_key = ApiKey(
|
||
key_hash=key_hash,
|
||
name=body.name,
|
||
permissions=body.permissions,
|
||
)
|
||
db.add(db_key)
|
||
await db.flush()
|
||
await db.refresh(db_key)
|
||
|
||
# Build response with plaintext key included (shown once)
|
||
base_data = ApiKeyResponse.model_validate(db_key).model_dump()
|
||
base_data["key"] = raw_key
|
||
response_data = ApiKeyCreateResponse(**base_data)
|
||
|
||
return APIResponse(
|
||
message="API key created. Store the key securely — it won't be shown again. / API密钥已创建，请妥善保管",
|
||
data=response_data,
|
||
)
|
||
|
||
|
||
@router.put(
|
||
"/{key_id}",
|
||
response_model=APIResponse[ApiKeyResponse],
|
||
summary="更新API密钥 / Update API key",
|
||
dependencies=[Depends(require_admin)],
|
||
)
|
||
async def update_key(
|
||
key_id: int, body: ApiKeyUpdate, db: AsyncSession = Depends(get_db)
|
||
):
|
||
"""更新API密钥的名称、权限或激活状态 / Update key name, permissions, or active status."""
|
||
result = await db.execute(select(ApiKey).where(ApiKey.id == key_id))
|
||
db_key = result.scalar_one_or_none()
|
||
if db_key is None:
|
||
raise HTTPException(status_code=404, detail="API key not found / 未找到密钥")
|
||
|
||
update_data = body.model_dump(exclude_unset=True)
|
||
for field, value in update_data.items():
|
||
setattr(db_key, field, value)
|
||
|
||
await db.flush()
|
||
await db.refresh(db_key)
|
||
return APIResponse(
|
||
message="API key updated / 密钥已更新",
|
||
data=ApiKeyResponse.model_validate(db_key),
|
||
)
|
||
|
||
|
||
@router.delete(
|
||
"/{key_id}",
|
||
response_model=APIResponse,
|
||
summary="停用API密钥 / Deactivate API key",
|
||
dependencies=[Depends(require_admin)],
|
||
)
|
||
async def deactivate_key(key_id: int, db: AsyncSession = Depends(get_db)):
|
||
"""停用API密钥（软删除）/ Deactivate an API key (soft delete)."""
|
||
result = await db.execute(select(ApiKey).where(ApiKey.id == key_id))
|
||
db_key = result.scalar_one_or_none()
|
||
if db_key is None:
|
||
raise HTTPException(status_code=404, detail="API key not found / 未找到密钥")
|
||
|
||
db_key.is_active = False
|
||
await db.flush()
|
||
return APIResponse(message="API key deactivated / 密钥已停用")
|