133 lines
4.1 KiB
Python
133 lines
4.1 KiB
Python
"""用户管理路由（超管）"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..database import get_db
|
|
from ..models import User, AuditLog
|
|
from ..security import hash_password
|
|
from ..permissions import ROLE_LABELS, PERMISSIONS, get_user_permissions
|
|
from ..schemas import UserCreate, UserUpdate, UserInfo
|
|
from ..deps import require_permission, get_current_user
|
|
|
|
router = APIRouter(prefix="/api/users", tags=["用户管理"])
|
|
|
|
|
|
@router.get("", response_model=list[UserInfo])
|
|
def list_users(
|
|
db: Session = Depends(get_db),
|
|
_: User = Depends(require_permission("user:view")),
|
|
):
|
|
users = db.query(User).order_by(User.id).all()
|
|
result = []
|
|
for u in users:
|
|
result.append(UserInfo(
|
|
id=u.id,
|
|
username=u.username,
|
|
role=u.role,
|
|
is_active=u.is_active,
|
|
remark=u.remark or "",
|
|
created_at=u.created_at,
|
|
permissions=get_user_permissions(u),
|
|
custom_permissions=u.custom_permissions,
|
|
))
|
|
return result
|
|
|
|
|
|
@router.post("", response_model=UserInfo)
|
|
def create_user(
|
|
req: UserCreate,
|
|
db: Session = Depends(get_db),
|
|
current: User = Depends(require_permission("user:create")),
|
|
):
|
|
if db.query(User).filter(User.username == req.username).first():
|
|
raise HTTPException(status_code=400, detail="用户名已存在")
|
|
|
|
user = User(
|
|
username=req.username,
|
|
password_hash=hash_password(req.password),
|
|
role=req.role,
|
|
remark=req.remark,
|
|
is_active=True,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
db.add(AuditLog(user_id=current.id, username=current.username,
|
|
action="user:create", target=user.username))
|
|
db.commit()
|
|
|
|
return UserInfo(
|
|
id=user.id, username=user.username, role=user.role,
|
|
is_active=user.is_active, remark=user.remark or "",
|
|
permissions=get_user_permissions(user),
|
|
custom_permissions=user.custom_permissions,
|
|
)
|
|
|
|
|
|
@router.put("/{user_id}", response_model=UserInfo)
|
|
def update_user(
|
|
user_id: int,
|
|
req: UserUpdate,
|
|
db: Session = Depends(get_db),
|
|
current: User = Depends(require_permission("user:edit")),
|
|
):
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
|
|
if req.password:
|
|
user.password_hash = hash_password(req.password)
|
|
if req.role is not None:
|
|
user.role = req.role
|
|
if req.is_active is not None:
|
|
user.is_active = req.is_active
|
|
if req.remark is not None:
|
|
user.remark = req.remark
|
|
if "custom_permissions" in req.model_fields_set:
|
|
user.custom_permissions = req.custom_permissions
|
|
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
db.add(AuditLog(user_id=current.id, username=current.username,
|
|
action="user:edit", target=user.username))
|
|
db.commit()
|
|
|
|
return UserInfo(
|
|
id=user.id, username=user.username, role=user.role,
|
|
is_active=user.is_active, remark=user.remark or "",
|
|
permissions=get_user_permissions(user),
|
|
custom_permissions=user.custom_permissions,
|
|
)
|
|
|
|
|
|
@router.delete("/{user_id}")
|
|
def delete_user(
|
|
user_id: int,
|
|
db: Session = Depends(get_db),
|
|
current: User = Depends(require_permission("user:delete")),
|
|
):
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
if user.role == "super_admin":
|
|
raise HTTPException(status_code=400, detail="不能删除超级管理员")
|
|
|
|
db.add(AuditLog(user_id=current.id, username=current.username,
|
|
action="user:delete", target=user.username))
|
|
db.delete(user)
|
|
db.commit()
|
|
return {"message": "已删除", "success": True}
|
|
|
|
|
|
@router.get("/permissions/list")
|
|
def list_permissions(_: User = Depends(require_permission("user:assign_permissions"))):
|
|
"""返回所有可用权限点及角色默认权限映射。"""
|
|
from ..permissions import ROLE_PERMISSIONS
|
|
return {
|
|
"permissions": PERMISSIONS,
|
|
"role_permissions": ROLE_PERMISSIONS,
|
|
}
|