simonkoson
106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
"""
|
|
用户管理 API — 管理员 CRUD（仅 zhipianren 可操作）
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlmodel import Session, select
|
|
|
|
from app.core.deps import get_current_user, require_role
|
|
from app.core.security import hash_password
|
|
from app.db.session import get_session
|
|
from app.models.user import User, UserRole
|
|
from app.schemas.user_admin import UserCreate, UserResponse, UserUpdate
|
|
|
|
router = APIRouter(prefix="/api/users", tags=["用户管理"])
|
|
|
|
|
|
@router.get("", response_model=list[UserResponse])
|
|
def list_users(
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(require_role(UserRole.zhipianren)),
|
|
):
|
|
"""获取所有用户列表（仅管理员）。"""
|
|
statement = select(User).order_by(User.id)
|
|
users = session.exec(statement).all()
|
|
return users
|
|
|
|
|
|
@router.get("/{user_id}", response_model=UserResponse)
|
|
def get_user(
|
|
user_id: int,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(require_role(UserRole.zhipianren)),
|
|
):
|
|
"""根据 ID 获取单个用户（仅管理员）。"""
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
|
|
return user
|
|
|
|
|
|
@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_user(
|
|
body: UserCreate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(require_role(UserRole.zhipianren)),
|
|
):
|
|
"""创建新用户（仅管理员）。"""
|
|
# 检查用户名唯一
|
|
existing = session.exec(select(User).where(User.username == body.username)).first()
|
|
if existing:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="用户名已存在")
|
|
|
|
user = User(
|
|
username=body.username,
|
|
display_name=body.display_name,
|
|
password_hash=hash_password(body.password),
|
|
role=body.role,
|
|
profile_text=body.profile_text,
|
|
)
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
|
|
@router.patch("/{user_id}", response_model=UserResponse)
|
|
def update_user(
|
|
user_id: int,
|
|
body: UserUpdate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(require_role(UserRole.zhipianren)),
|
|
):
|
|
"""更新用户信息（仅管理员）。可更新 display_name/role/profile_text/is_active。"""
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
|
|
|
|
update_data = body.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(user, field, value)
|
|
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
|
|
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def deactivate_user(
|
|
user_id: int,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(require_role(UserRole.zhipianren)),
|
|
):
|
|
"""停用用户（软删除，仅管理员）。"""
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
|
|
|
|
# 不能停用自己的账号
|
|
if user.id == current_user.id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="不能停用自己的账号")
|
|
|
|
user.is_active = False
|
|
session.add(user)
|
|
session.commit()
|
|
return None |