86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
||
from sqlalchemy import select
|
||
|
||
from app.core.database import async_session
|
||
from app.models.api_key import ApiKeyModel
|
||
from app.api.deps import require_admin, AuthContext
|
||
|
||
# Все операции с ключами -- только для админов (мастер-ключ)
|
||
router = APIRouter(dependencies=[Depends(require_admin)])
|
||
|
||
|
||
@router.get("")
|
||
async def list_keys():
|
||
"""Список всех гостевых ключей."""
|
||
async with async_session() as session:
|
||
result = await session.execute(select(ApiKeyModel))
|
||
keys = result.scalars().all()
|
||
|
||
return [
|
||
{
|
||
"key": k.key,
|
||
"name": k.name,
|
||
"is_admin": k.is_admin,
|
||
"active": k.active,
|
||
"created_at": k.created_at,
|
||
}
|
||
for k in keys
|
||
]
|
||
|
||
|
||
@router.post("")
|
||
async def create_key(name: str, is_admin: bool = False):
|
||
"""Создать гостевой ключ. Возвращает сгенерированный токен."""
|
||
new_key = ApiKeyModel(
|
||
key=ApiKeyModel.generate_key(),
|
||
name=name,
|
||
is_admin=is_admin,
|
||
)
|
||
|
||
async with async_session() as session:
|
||
session.add(new_key)
|
||
await session.commit()
|
||
|
||
return {
|
||
"key": new_key.key,
|
||
"name": new_key.name,
|
||
"is_admin": new_key.is_admin,
|
||
"message": "Сохраните ключ -- он больше не будет показан полностью",
|
||
}
|
||
|
||
|
||
@router.delete("/{key}")
|
||
async def revoke_key(key: str):
|
||
"""Деактивировать (отозвать) гостевой ключ."""
|
||
async with async_session() as session:
|
||
result = await session.execute(
|
||
select(ApiKeyModel).where(ApiKeyModel.key == key)
|
||
)
|
||
api_key = result.scalar_one_or_none()
|
||
if not api_key:
|
||
raise HTTPException(status_code=404, detail="Ключ не найден")
|
||
|
||
api_key.active = False
|
||
session.add(api_key)
|
||
await session.commit()
|
||
|
||
return {"status": "revoked", "name": api_key.name}
|
||
|
||
|
||
@router.post("/{key}/activate")
|
||
async def activate_key(key: str):
|
||
"""Повторно активировать ключ."""
|
||
async with async_session() as session:
|
||
result = await session.execute(
|
||
select(ApiKeyModel).where(ApiKeyModel.key == key)
|
||
)
|
||
api_key = result.scalar_one_or_none()
|
||
if not api_key:
|
||
raise HTTPException(status_code=404, detail="Ключ не найден")
|
||
|
||
api_key.active = True
|
||
session.add(api_key)
|
||
await session.commit()
|
||
|
||
return {"status": "activated", "name": api_key.name}
|