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": "用户已删除"}
|