83 lines
3.4 KiB
Python
83 lines
3.4 KiB
Python
"""
|
|
===============================================================================
|
|
FILE: modules/web_api/routers/admin.py
|
|
ROLE: Администрирование пользователей и прав доступа.
|
|
===============================================================================
|
|
"""
|
|
|
|
# ANCHOR[ADMIN_ROUTER_IMPORTS]
|
|
import logging
|
|
from typing import Dict, Any, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from .auth import get_current_user, get_db, pwd_context
|
|
|
|
router = APIRouter(prefix="/api/v1/admin", tags=["admin"])
|
|
|
|
# ANCHOR[ADMIN_SCHEMAS]
|
|
class CreateUserRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
full_name: Optional[str] = None
|
|
is_admin: Optional[bool] = False
|
|
|
|
# ANCHOR[ADMIN_ENDPOINTS]
|
|
@router.get("/users")
|
|
def list_users(current_user: Dict[str, Any] = Depends(get_current_user)):
|
|
if not current_user["is_admin"]:
|
|
raise HTTPException(status_code=403, detail="Доступ запрещен. Только для администратора.")
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT id, username, full_name, is_admin, created_at FROM users ORDER BY id ASC")
|
|
users = [dict(r) for r in cursor.fetchall()]
|
|
conn.close()
|
|
return users
|
|
|
|
@router.post("/users")
|
|
def create_user(req: CreateUserRequest, current_user: Dict[str, Any] = Depends(get_current_user)):
|
|
if not current_user["is_admin"]:
|
|
raise HTTPException(status_code=403, detail="Доступ запрещен. Только для администратора.")
|
|
|
|
username = req.username.strip().lower()
|
|
if not username or not req.password:
|
|
raise HTTPException(status_code=400, detail="Заполните имя пользователя и пароль")
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT id FROM users WHERE username = ?", (username,))
|
|
if cursor.fetchone():
|
|
conn.close()
|
|
raise HTTPException(status_code=400, detail="Пользователь с таким именем уже существует")
|
|
|
|
pwd_hash = pwd_context.hash(req.password)
|
|
full_name = req.full_name.strip() if req.full_name else None
|
|
is_admin = 1 if req.is_admin else 0
|
|
|
|
cursor.execute(
|
|
"INSERT INTO users (username, password_hash, full_name, is_admin) VALUES (?, ?, ?, ?)",
|
|
(username, pwd_hash, full_name, is_admin)
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
logging.info(f"Создан пользователь: {username} (admin={is_admin}) админом {current_user['username']}")
|
|
return {"status": "success", "message": f"Пользователь {username} создан"}
|
|
|
|
@router.delete("/users/{user_id}")
|
|
def delete_user(user_id: int, current_user: Dict[str, Any] = Depends(get_current_user)):
|
|
if not current_user["is_admin"]:
|
|
raise HTTPException(status_code=403, detail="Доступ запрещен. Только для администратора.")
|
|
|
|
if user_id == current_user["id"]:
|
|
raise HTTPException(status_code=400, detail="Нельзя удалить самого себя")
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM users WHERE id = ?", (user_id,))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
logging.info(f"Удален пользователь ID: {user_id}")
|
|
return {"status": "success", "message": "Пользователь удален"} |