admin/admin.py - master - dmitry.kovalenkov/webcc

admin/admin.py
132 строки6 KB
dmkovalenkov
update
27 апр 2026, 21:13
27 апр 2026, 21:13e6bfd88
from typing import List, Optional
from fastapi import APIRouter, HTTPException, Depends, Query
import db
import schemas
import security

router = APIRouter()


# ──────────────────────────────────────────────
#  GET /admin/groups
#  Список всех групп
# ──────────────────────────────────────────────
@router.get("/groups", response_model=List[schemas.GroupOut])
def list_groups(payload=Depends(security.require_permission("groups:read"))):
    return [schemas.GroupOut(**g) for g in db.list_groups()]


# ──────────────────────────────────────────────
#  POST /admin/groups
#  Создать группу
# ──────────────────────────────────────────────
@router.post("/groups", response_model=schemas.GroupOut, status_code=201)
def create_group(
    body: schemas.GroupCreate,
    payload=Depends(security.require_permission("groups:create")),
):
    group = db.create_group(name=body.name, description=body.description)
    return schemas.GroupOut(**group)


# ──────────────────────────────────────────────
#  PUT /admin/groups/{id}
#  Обновить название / описание группы
# ──────────────────────────────────────────────
@router.put("/groups/{group_id}", response_model=schemas.GroupOut)
def update_group(
    group_id: int,
    body: schemas.GroupUpdate,
    payload=Depends(security.require_permission("groups:update")),
):
    existing = db.get_group_by_id(group_id)
    if existing is None:
        raise HTTPException(status_code=404, detail="Group not found")
    fields = body.model_dump(exclude_none=True)
    updated = db.update_group(group_id, fields)
    return schemas.GroupOut(**updated)


# ──────────────────────────────────────────────
#  GET /admin/groups/{id}/members
#  Состав группы
# ──────────────────────────────────────────────
@router.get("/groups/{group_id}/members")
def get_group_members(
    group_id: int,
    payload=Depends(security.require_permission("groups:read")),
):
    existing = db.get_group_by_id(group_id)
    if existing is None:
        raise HTTPException(status_code=404, detail="Group not found")
    return db.get_group_members(group_id)


# ──────────────────────────────────────────────
#  PUT /admin/groups/{id}/members
#  Полная замена состава группы
#  Body: {"members": [{"user_id": 2, "role_in_group": "operator"}, ...]}
# ──────────────────────────────────────────────
@router.put("/groups/{group_id}/members")
def set_group_members(
    group_id: int,
    body: schemas.GroupMembersUpdate,
    payload=Depends(security.require_permission("groups:manage_members")),
):
    existing = db.get_group_by_id(group_id)
    if existing is None:
        raise HTTPException(status_code=404, detail="Group not found")

    for m in body.members:
        if m.get("role_in_group") not in ("operator", "supervisor"):
            raise HTTPException(status_code=400, detail=f"Invalid role_in_group for user_id={m.get('user_id')}")

    db.set_group_members(group_id, body.members)
    return {"group_id": group_id, "members": body.members}


# ──────────────────────────────────────────────
#  GET /admin/groups/{id}/scenarios
#  Сценарии, назначенные группе
# ──────────────────────────────────────────────
@router.get("/groups/{group_id}/scenarios")
def get_group_scenarios(
    group_id: int,
    payload=Depends(security.require_permission("groups:read")),
):
    existing = db.get_group_by_id(group_id)
    if existing is None:
        raise HTTPException(status_code=404, detail="Group not found")
    return db.get_group_scenarios(group_id)


# ──────────────────────────────────────────────
#  PUT /admin/groups/{id}/scenarios
#  Полная замена списка сценариев группы
#  Body: {"scenario_ids": [1, 2, 3]}
# ──────────────────────────────────────────────
@router.put("/groups/{group_id}/scenarios")
def set_group_scenarios(
    group_id: int,
    body: schemas.GroupScenariosUpdate,
    payload=Depends(security.require_permission("groups:manage_scenarios")),
):
    existing = db.get_group_by_id(group_id)
    if existing is None:
        raise HTTPException(status_code=404, detail="Group not found")
    db.set_group_scenarios(group_id, body.scenario_ids)
    return {"group_id": group_id, "scenario_ids": body.scenario_ids}


# ──────────────────────────────────────────────
#  GET /admin/users
#  Список пользователей (чтение из USERS)
# ──────────────────────────────────────────────
@router.get("/users", response_model=List[schemas.UserOut])
def list_users(
    role: Optional[str] = Query(None),
    status: Optional[str] = Query(None),
    payload=Depends(security.require_permission("users:read")),
):
    users = db.list_users(role=role, status=status)
    return [schemas.UserOut(**u) for u in users]