ajyy/backend/app/crud/user.py

"""用户 CRUD"""
from typing import List, Optional, Tuple

from sqlalchemy.orm import Session

from app.models.system import SysRole, SysUser


def get_user_by_username(db: Session, username: str) -> Optional[SysUser]:
    return (
        db.query(SysUser)
        .filter(SysUser.username == username, SysUser.del_flag == "0")
        .first()
    )


def get_user_by_id(db: Session, user_id: str) -> Optional[SysUser]:
    return (
        db.query(SysUser)
        .filter(SysUser.id == user_id, SysUser.del_flag == "0")
        .first()
    )


def get_role_by_id(db: Session, role_id: str) -> Optional[SysRole]:
    return (
        db.query(SysRole)
        .filter(SysRole.id == role_id, SysRole.del_flag == "0")
        .first()
    )


def get_role_by_code(db: Session, role_code: str) -> Optional[SysRole]:
    return (
        db.query(SysRole)
        .filter(SysRole.role_code == role_code, SysRole.del_flag == "0")
        .first()
    )


def list_users(db: Session) -> List[SysUser]:
    return (
        db.query(SysUser)
        .filter(SysUser.del_flag == "0")
        .order_by(SysUser.create_time.desc())
        .all()
    )


def create_user(
    db: Session,
    *,
    username: str,
    password_hash: str,
    display_name: str,
    role_id: str,
    status: str = "0",
) -> SysUser:
    row = SysUser(
        username=username,
        password_hash=password_hash,
        display_name=display_name,
        role_id=role_id,
        status=status,
        del_flag="0",
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    return row


def update_user(
    db: Session,
    user: SysUser,
    *,
    display_name: Optional[str] = None,
    role_id: Optional[str] = None,
    status: Optional[str] = None,
    password_hash: Optional[str] = None,
) -> SysUser:
    if display_name is not None:
        user.display_name = display_name
    if role_id is not None:
        user.role_id = role_id
    if status is not None:
        user.status = status
    if password_hash is not None:
        user.password_hash = password_hash
    db.commit()
    db.refresh(user)
    return user


def get_user_with_role(
    db: Session, user_id: str
) -> Tuple[Optional[SysUser], Optional[SysRole]]:
    user = get_user_by_id(db, user_id)
    if not user:
        return None, None
    role = get_role_by_id(db, user.role_id)
    return user, role