python-fastapi-mm-zhcs-yj-api/routers/prod_api/system/role/__init__.py

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

from fastapi import APIRouter, Request, Depends,Query,HTTPException
from database import get_db
from sqlalchemy.orm import Session
from fastapi.responses import JSONResponse
from models import *
from utils import *
from utils.ry_system_util import *
from common.security import valid_access_token
import traceback

router = APIRouter()


@router.post('')
async def rolecreate(
                    db: Session = Depends(get_db),
                    user_id: int = Depends(valid_access_token),
                    body = Depends(remove_xss_json)
):
    try:


        role.role_key = body['roleKey']
        role.role_name = body['roleName']
        role.role_sort = body['roleSort']
        role.status = body['status']
        role.remark = body['remark']
        menuCheckStrictly = body['menuCheckStrictly']
        if menuCheckStrictly:
            role.menu_check_strictly = 1
        else:
            role.menu_check_strictly = 0

        menuIds = body['menuIds']


        # 清除当前用户的所有角色关联
        db.query(SysRoleMenu).filter(SysRoleMenu.role_id == roleId).delete()

        # 创建新的用户角色关联
        new_role_menus = [SysRoleMenu(role_id=roleId, menu_id=menu_id) for menu_id in menuIds]
        db.add_all(new_role_menus)


        db.commit()
        return {"code": 200, "msg": "更新成功", "data": None}


    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@router.put('')
async def roleupdate(
                    db: Session = Depends(get_db),
                    user_id: int = Depends(valid_access_token),
                    body = Depends(remove_xss_json)
):
    try:
        roleId = body['roleId']

        query = db.query(SysRole)
        query = query.filter(SysRole.del_flag != '2')
        query = query.filter(SysRole.role_id == roleId)
        role = query.first()
        if not role :
            return JSONResponse(status_code=410, content={
            'errcode': 410,
            'errmsg': f'角色{roleId}不存在'
        })

        role.role_key = body['roleKey']
        role.role_name = body['roleName']
        role.role_sort = body['roleSort']
        role.status = body['status']
        role.remark = body['remark']
        menuCheckStrictly = body['menuCheckStrictly']
        if menuCheckStrictly:
            role.menu_check_strictly = 1
        else:
            role.menu_check_strictly = 0

        menuIds = body['menuIds']


        # 清除当前用户的所有角色关联
        db.query(SysRoleMenu).filter(SysRoleMenu.role_id == roleId).delete()

        # 创建新的用户角色关联
        new_role_menus = [SysRoleMenu(role_id=roleId, menu_id=menu_id) for menu_id in menuIds]
        db.add_all(new_role_menus)


        db.commit()
        return {"code": 200, "msg": "更新成功", "data": None}


    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")





@router.get('/list')
async def rolelist( roleName: int = Query(None ,description='角色名称'),
                    roleKey: str = Query(None, description='权限字符'),
                    status: int = Query(None, description='角色状态'),
                    # beginTime: str = Depends(get_time_params),
                    # endTime: str = Depends(get_time_params),
                    page: int = Query(1, gt=0, description='页码'),
                    pageSize: int = Query(10, gt=0, description='每页条目数量'),
                    db: Session = Depends(get_db),
                    user_id: int = Depends(valid_access_token)):
    try:
        # 构建查询
        query = db.query(SysRole)
        query = query.filter(SysRole.del_flag != '2')
        # 应用查询条件

        # if beginTime:
        #     query = query.filter(SysUser.create_time >= beginTime)
        # if endTime:
        #     query = query.filter(SysUser.create_time <= endTime)
        # if params:
        #     query = query.filter(SysUser.create_time >= params.get("beginTime"))
        #     query = query.filter(SysUser.create_time <= params.get("endTime"))
        if roleName:
            query =query.filter(SysRole.role_name.like(f'%{roleName}%'))
        if status:
            query =query.filter(SysRole.status==status)
        if roleKey:
            query =query.filter(SysRole.role_key.like(f'%{roleKey}%'))

        # 计算总条目数
        total_items = query.count()

        # 排序
        query = query.order_by(SysRole.create_time.desc())
        # 执行分页查询
        roles = query.offset((page - 1) * pageSize).limit(pageSize).all()

        # 将查询结果转换为列表形式的字典
        role_list = role_list_to_dict(roles,[])


        # 返回结果
        return {
            "code": 200,
            "msg": "成功",
            "rows": role_list,
            "total": total_items,
            "page": page,
            "pageSize": pageSize,
            "totalPages": (total_items + pageSize - 1) // pageSize
        }

    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@router.get('/{roleId}')
async def rolelist( roleId:int,
                    db: Session = Depends(get_db),
                    user_id: int = Depends(valid_access_token)):
    try:
        # 构建查询
        query = db.query(SysRole)
        query = query.filter(SysRole.del_flag != '2')
        # 应用查询条件
        query = query.filter(SysRole.role_id==roleId)

        roles = query.all()

        # 将查询结果转换为列表形式的字典
        role_list = role_list_to_dict(roles,[])

        # 返回结果
        return {
            "code": 200,
            "msg": "成功",
            "data": role_list[0]
        }

    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")