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.sysdata.role import crud_role
from crud.user.user import crud_admin
from models.user import Admin
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_admin(user: CreateAdminUser,
                       db: AsyncSession = Depends(get_async_db),
                       current_user: Admin = Depends(get_current_user)):
    user_in = user.dict(exclude_none=True)
    if not user:
        return {"errcode": 400, "mess": "请求参数为空!"}

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

    user_in["creator_id"] = current_user.id
    user_in["creator_name"] = current_user.username
    user_in["editor_id"] = current_user.id
    user_in["editor_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_admin(user: UpdateAdminUser,
                       uid: int = Path(..., description="管理员ID"),
                       db: AsyncSession = Depends(get_async_db),
                       current_user: Admin = Depends(get_current_user)):
    user_in = user.dict(exclude_none=True)
    if not user:
        return {"errcode": 400, "mess": "请求参数为空!"}

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


@router.delete("/sysuser/{uid}", summary="删除管理员", response_model=AdminUserDetail)
async def delete_admin(uid: int = Path(..., description="管理员ID"),
                       db: AsyncSession = Depends(get_async_db),
                       current_user: Admin = Depends(get_current_user)):
    # 不允许删除初始管理员账号
    if uid == 1:
        return {"errcode": 400, "mess": "该用户不允许删除!"}

    existed = await crud_admin.count(db, filters={"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_admin(uid: int = Path(..., description="管理员ID"),
                    db: AsyncSession = Depends(get_async_db),
                    current_user: Admin = Depends(get_current_user)):
    db_obj = await crud_admin.find_one(db, filters={"id": uid})
    if not db_obj:
        return {"errcode": 404, "mess": "管理员不存在!"}
    else:
        return {"data": db_obj}


@router.get("/sysusers",
            summary="获取管理员列表",
            response_model=AdminUserList,
            response_model_exclude_none=True)
async def get_admin_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: Admin = Depends(get_current_user)):
    _q = []
    if username:
        _q.append(Admin.username.like(f"%{username}%"))
    if name:
        _q.append(Admin.name.like(f"%{name}%"))
    if phone:
        _q.append(Admin.phone.like(f"%{phone}%"))
    offset = (page - 1) * size

    if isinstance(order, str):
        order = [text(order)]
    total, items = await crud_admin.find_all(db,
                                             filters=_q,
                                             offset=offset,
                                             limit=size,
                                             order_by=order,
                                             return_fields=res)
    for item in items:
        print(item.role_id)
        role = await crud_role.find_one(db, filters={"id": item.role_id})
        item.role_name = role.name
    return {"total": total, "data": items}