xzbautomarkapi/admin/api/endpoints/sysdata/user.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# 系统用户管理
from fastapi import APIRouter, Depends, Path, Query
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import or_, and_

from core.security import hashed_password
from crud.user.user import crud_admin
from models.user import SysUser
from schemas.base import ReturnField, OrderByField
from schemas.user import CreateAdminUser, AdminUserDetail, UpdateAdminUser, AdminUserList
from utils.depends import get_async_db, get_current_user

router = APIRouter()


@router.post("/sysuser", summary="新增管理员", response_model=AdminUserDetail)
async def create_sysuser(user: CreateAdminUser,
                         db: AsyncSession = Depends(get_async_db),
                         current_user: SysUser = Depends(get_current_user)):
    user_in = user.dict(exclude_none=True)
    if not user:
        return {"errcode": 400, "mess": "请求参数为空!"}

    # 判断是否存在同名
    filters = [
        or_(SysUser.username == user_in["username"],
            SysUser.phone == user_in["username"])
    ]
    db_user = await crud_admin.find_one(db, filters)
    if db_user:
        return {"errcode": 400, "mess": "账号或手机号已存在!"}

    user_in["creator_id"] = current_user.id
    user_in["creator_name"] = current_user.username
    user_in["password"] = hashed_password(user.password)
    db_user = await crud_admin.insert_one(db, user_in)
    return {"data": db_user}


@router.put("/sysuser/{uid}", summary="修改管理员", response_model=AdminUserDetail)
async def update_sysuser(user: UpdateAdminUser,
                         uid: int = Path(..., description="管理员ID"),
                         db: AsyncSession = Depends(get_async_db),
                         current_user: SysUser = Depends(get_current_user)):
    user_in = user.dict(exclude_none=True)
    if not user:
        return {"errcode": 400, "mess": "请求参数为空!"}

    # 判断是否存在同名
    filters = [and_(SysUser.id != uid, SysUser.phone == user_in["phone"])]
    db_user = await crud_admin.find_one(db, filters)
    if db_user:
        return {"errcode": 400, "mess": "账号或手机号已存在!"}
    db_obj = await crud_admin.find_one(db, {"id": uid})

    user_in["editor_id"] = current_user.id
    user_in["editor_name"] = current_user.username
    db_user = await crud_admin.update(db, db_obj, user_in)
    return {"data": db_user}


@router.delete("/sysuser/{uid}",
               summary="删除管理员",
               response_model=AdminUserDetail)
async def delete_sysuser(uid: int = Path(..., description="管理员ID"),
                         db: AsyncSession = Depends(get_async_db),
                         current_user: SysUser = Depends(get_current_user)):
    existed = await crud_admin.count(db, {"id": uid})
    if not existed:
        return {"errcode": 404, "mess": "管理员不存在!"}
    else:
        await crud_admin.delete(db, obj_id=uid)
        return {"data": None}


@router.get("/sysuser/{uid}", summary="获取管理员详情", response_model=AdminUserDetail)
async def get_sysuser(uid: int = Path(..., description="管理员ID"),
                      db: AsyncSession = Depends(get_async_db),
                      current_user: SysUser = Depends(get_current_user)):
    existed = await crud_admin.count(db, {"id": uid})
    if not existed:
        return {"errcode": 404, "mess": "管理员不存在!"}
    else:
        db_obj = await crud_admin.find_one(db, {"id": uid})
        return {"data": db_obj}


@router.get("/sysusers",
            summary="获取管理员列表",
            response_model=AdminUserList,
            response_model_exclude_none=True)
async def get_sysuser_list(
    page: int = 1,
    size: int = 10,
    username: str = "",
    name: str = "",
    phone: str = "",
    order: OrderByField = Query("-created_at",
                                description="排序字段,用逗号分隔,升降序以-判断,默认-created_at"),
    res: ReturnField = Query("",
                             description="返回字段，取值范围：id,username,name,phone"),
    db: AsyncSession = Depends(get_async_db),
    current_user: SysUser = Depends(get_current_user)):
    filters = []
    if username:
        filters.append(text(f"username Like '%{username}%'"))
    if name:
        filters.append(text(f"name LIKE '%{name}%'"))
    if phone:
        filters.append(text(f"phone LIKE '%{phone}%'"))
    offset = (page - 1) * size

    if isinstance(order, str):
        order = [text(order)]
    total, items = await crud_admin.find_all(db,
                                             filters=filters,
                                             offset=offset,
                                             limit=size,
                                             order_by=order,
                                             return_fields=res)
    return {"total": total, "data": items}