100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
"""用户管理 API（仅管理员）"""
|
|
import uuid
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..database import get_db
|
|
from ..models import User
|
|
from ..deps import get_current_user
|
|
from ..security import hash_password
|
|
|
|
router = APIRouter(prefix="/api/users", tags=["users"], dependencies=[Depends(get_current_user)])
|
|
|
|
|
|
def _require_admin(user: dict):
|
|
if not user.get("is_admin"):
|
|
raise HTTPException(status_code=403, detail="需要管理员权限")
|
|
|
|
|
|
@router.get("")
|
|
async def list_users(user: dict = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
"""获取用户列表"""
|
|
_require_admin(user)
|
|
users = db.query(User).order_by(User.created_at.desc()).all()
|
|
return [{"id": u.id, "username": u.username, "is_admin": u.is_admin, "created_at": str(u.created_at) if u.created_at else None} for u in users]
|
|
|
|
|
|
@router.post("")
|
|
async def create_user(data: dict, user: dict = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
"""创建用户"""
|
|
_require_admin(user)
|
|
|
|
username = data.get("username", "").strip()
|
|
password = data.get("password", "").strip()
|
|
is_admin = data.get("is_admin", False)
|
|
|
|
if not username or not password:
|
|
raise HTTPException(status_code=400, detail="用户名和密码不能为空")
|
|
if len(password) < 11:
|
|
raise HTTPException(status_code=400, detail="密码至少 11 位")
|
|
if db.query(User).filter(User.username == username).first():
|
|
raise HTTPException(status_code=400, detail="用户名已存在")
|
|
|
|
new_user = User(
|
|
id=str(uuid.uuid4()),
|
|
username=username,
|
|
password_hash=hash_password(password),
|
|
is_admin=is_admin,
|
|
)
|
|
db.add(new_user)
|
|
db.commit()
|
|
return {"message": "用户已创建", "id": new_user.id}
|
|
|
|
|
|
@router.put("/{user_id}/password")
|
|
async def change_password(user_id: str, data: dict, user: dict = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
"""修改密码（管理员可改任意用户）"""
|
|
_require_admin(user)
|
|
|
|
target = db.query(User).filter(User.id == user_id).first()
|
|
if not target:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
|
|
new_password = data.get("password", "").strip()
|
|
if not new_password or len(new_password) < 11:
|
|
raise HTTPException(status_code=400, detail="密码至少 11 位")
|
|
|
|
target.password_hash = hash_password(new_password)
|
|
db.commit()
|
|
return {"message": "密码已更新"}
|
|
|
|
|
|
@router.put("/{user_id}/role")
|
|
async def change_role(user_id: str, data: dict, user: dict = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
"""切换管理员状态"""
|
|
_require_admin(user)
|
|
|
|
target = db.query(User).filter(User.id == user_id).first()
|
|
if not target:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
|
|
target.is_admin = data.get("is_admin", not target.is_admin)
|
|
db.commit()
|
|
return {"message": "角色已更新", "is_admin": target.is_admin}
|
|
|
|
|
|
@router.delete("/{user_id}")
|
|
async def delete_user(user_id: str, user: dict = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
"""删除用户"""
|
|
_require_admin(user)
|
|
|
|
target = db.query(User).filter(User.id == user_id).first()
|
|
if not target:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
if target.username == user["username"]:
|
|
raise HTTPException(status_code=400, detail="不能删除自己")
|
|
|
|
db.delete(target)
|
|
db.commit()
|
|
return {"message": "用户已删除"}
|