iTi-System/iti_system/routes/user.py

from __future__ import annotations

from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload

from iti.auth import require_permission
from iti.db import get_db
from iti.exceptions import BizError
from iti.responses import ok, page

from iti_system.models import Role, User
from iti_system.schemas import PasswordUpdate, UserCreate, UserQuery, UserUpdate
from iti_system.services import (
    apply_keyword,
    assert_unique_user,
    bind_depts,
    bind_roles,
    current_user,
    dump_user,
    get_or_404,
    paginate_query,
)


router = APIRouter(prefix="/sys/user", tags=["system.user"])


@router.get("/current")
def get_current_user(user: User = Depends(current_user)):
    return ok(dump_user(user))


@router.get("/list", dependencies=[Depends(require_permission("system:user:list"))])
def list_user(query: UserQuery = Depends(), db: Session = Depends(get_db)):
    stmt = _query_users(query)
    return ok([dump_user(user) for user in db.scalars(stmt).all()])


@router.get("/page", dependencies=[Depends(require_permission("system:user:list"))])
def page_user(query: UserQuery = Depends(), db: Session = Depends(get_db)):
    result = paginate_query(db, _query_users(query), page=query.page, size=query.size)
    return page([dump_user(user) for user in result["items"]], result["page"])


@router.post("", dependencies=[Depends(require_permission("system:user:create"))])
def create_user(payload: UserCreate, db: Session = Depends(get_db)):
    assert_unique_user(
        db,
        username=payload.username,
        phone=payload.phone,
        email=payload.email,
    )
    user = User(
        username=payload.username,
        phone=payload.phone,
        email=payload.email,
        realname=payload.realname,
        desc=payload.desc,
        avatar=payload.avatar,
        gender=payload.gender,
        status=payload.status,
        roles=bind_roles(db, payload.roles),
        depts=bind_depts(db, payload.depts),
    )
    user.set_password(payload.password or "123456")
    db.add(user)
    db.commit()
    db.refresh(user)
    return ok(dump_user(user))


@router.put("/password", dependencies=[Depends(require_permission("system:user:resetpwd"))])
def update_password(payload: PasswordUpdate, user: User = Depends(current_user), db: Session = Depends(get_db)):
    if payload.type == "updateByOldPassword" and not user.check_password(payload.old_password or ""):
        raise BizError("密码错误")
    user.set_password(payload.new_password)
    db.commit()
    return ok()


@router.put("/{id}", dependencies=[Depends(require_permission("system:user:edit"))])
def update_user(id: str, payload: UserUpdate, db: Session = Depends(get_db)):
    user = get_or_404(db, User, id, "用户不存在")
    values = payload.model_dump(exclude_unset=True)
    assert_unique_user(
        db,
        user_id=id,
        username=values.get("username"),
        phone=values.get("phone"),
        email=values.get("email"),
    )
    for key in ("username", "phone", "email", "realname", "desc", "avatar", "gender", "status"):
        if key in values:
            setattr(user, key, values[key])
    if values.get("roles") is not None:
        user.roles = bind_roles(db, values["roles"])
    if values.get("depts") is not None:
        user.depts = bind_depts(db, values["depts"])
    db.commit()
    db.refresh(user)
    return ok(dump_user(user))


@router.delete("/{id}", dependencies=[Depends(require_permission("system:user:delete"))])
def delete_user(id: str, db: Session = Depends(get_db)):
    user = get_or_404(db, User, id, "用户不存在")
    db.delete(user)
    db.commit()
    return ok()


def _query_users(query: UserQuery):
    stmt = (
        select(User)
        .options(
            selectinload(User.roles).selectinload(Role.menus),
            selectinload(User.depts),
            selectinload(User.attributes),
        )
        .order_by(User.created_at.desc())
    )
    stmt = apply_keyword(stmt, query.keyword, User.username, User.phone, User.email, User.realname)
    if query.username:
        stmt = stmt.where(User.username.like(f"%{query.username}%"))
    if query.phone:
        stmt = stmt.where(User.phone.like(f"%{query.phone}%"))
    if query.email:
        stmt = stmt.where(User.email.like(f"%{query.email}%"))
    if query.realname:
        stmt = stmt.where(User.realname.like(f"%{query.realname}%"))
    if query.gender:
        stmt = stmt.where(User.gender == query.gender)
    if query.status:
        stmt = stmt.where(User.status == query.status)
    return stmt