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

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

from typing import Optional

from fastapi import APIRouter, Depends, Query
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from crud.sysdata.role import crud_role, crud_permission
from models.user import SysUser
from schemas.base import ReturnField, OrderByField
from schemas.sysdata.role import RoleList, RoleInfo, RoleDetail, PermissionList
from utils.depends import get_async_db, get_current_user

router = APIRouter()


@router.post("/role", summary="新增角色", response_model=RoleDetail)
async def create_role(info: RoleInfo,
                      db: AsyncSession = Depends(get_async_db),
                      current_user: SysUser = Depends(get_current_user)):
    info = info.dict(exclude_none=True)
    if not info:
        return {"errcode": 400, "mess": "请求参数为空!"}
    db_exist = await crud_role.find_one(db, {"name": info["name"]})
    if db_exist:
        return {"errcode": 400, "mess": "角色名称重复!"}
    info["creator_id"] = current_user.id
    info["creator_name"] = current_user.username
    db_obj = await crud_role.insert_one(db, info)

    return {"data": db_obj}


@router.put("/role/{rid}", summary="修改角色", response_model=RoleDetail)
async def update_role(rid: int,
                      info: RoleInfo,
                      db: AsyncSession = Depends(get_async_db),
                      current_user: SysUser = Depends(get_current_user)):
    info = info.dict(exclude_none=True)
    if not info:
        return {"errcode": 400, "mess": "请求参数为空!"}
    db_exist = await crud_role.find_one(db, {"id": rid})
    if not db_exist:
        return {"errcode": 400, "mess": "角色不存在!"}
    info["editor_id"] = current_user.id
    info["editor_name"] = current_user.username
    db_obj = await crud_role.update(db, db_exist, info)

    return {"data": db_obj}


@router.delete("/role/{rid}", summary="删除角色", response_model=RoleDetail)
async def delete_role(rid: int,
                      db: AsyncSession = Depends(get_async_db),
                      current_user: SysUser = Depends(get_current_user)):
    existed = await crud_role.count(db, {"id": rid})
    if not existed:
        return {"errcode": 400, "mess": "角色不存在!"}
    await crud_role.delete(db, rid)
    return {"data": None}


@router.get("/role/{rid}", summary="角色详情", response_model=RoleDetail)
async def get_role(rid: int,
                   db: AsyncSession = Depends(get_async_db),
                   current_user: SysUser = Depends(get_current_user)):
    db_exist = await crud_role.find_one(db, {"id": rid})
    if not db_exist:
        return {"errcode": 400, "mess": "角色不存在!"}
    return {"data": db_exist}


@router.get("/roles", summary="角色列表", response_model=RoleList)
async def get_role_list(page: Optional[int] = None,
                        size: Optional[int] = None,
                        name: str = "",
                        res: ReturnField = Query("",
                                                 description="控制返回字段,字段逗号分隔"),
                        order: OrderByField = Query(
                            "-created_at",
                            description="排序字段,用逗号分隔,升降序以-判断,默认-created_at"),
                        db: AsyncSession = Depends(get_async_db),
                        current_user: SysUser = Depends(get_current_user)):
    filters = []
    if name:
        filters.append(text(f"name LIKE '%{name}%'"))
    if ((page is not None) and page >= 1) and ((size is not None)
                                               and size >= 1):
        offset = (page - 1) * size
    else:
        size = offset = None

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


@router.get("/permissions", summary="权限类别", response_model=PermissionList)
async def get_permission_all(name: str = "",
                             res: ReturnField = Query(
                                 "", description="控制返回字段,字段逗号分隔"),
                             db: AsyncSession = Depends(get_async_db),
                             current_user: SysUser = Depends(get_current_user)):
    filters = []
    if name:
        filters.append(text(f"name LIKE '%{name}%'"))

    items = await crud_permission.fetch_all(db,
                                            filters=filters,
                                            return_fields=res)
    return {"data": items}