# =========================================== # 班级操行分管理系统 - 管理员角色模型 # # 开发者: Canglan # 联系方式: admin@sea-studio.top # 版权归属: Sea Network Technology Studio # 许可证: MIT License # # 版权所有 © Sea Network Technology Studio # =========================================== from typing import Optional, Dict, Any, List from utils.database import execute_one, execute_query, execute_insert, execute_update class AdminRoleModel: """管理员角色数据模型""" @staticmethod async def get_by_user_id(user_id: int) -> Optional[Dict[str, Any]]: sql = """ SELECT ar.*, s.subject_name FROM admin_roles ar LEFT JOIN subjects s ON ar.subject_id = s.subject_id WHERE ar.user_id = %s LIMIT 1 """ return await execute_one(sql, (user_id,)) @staticmethod async def get_all() -> List[Dict[str, Any]]: sql = """ SELECT ar.*, u.real_name, u.username, s.subject_name FROM admin_roles ar JOIN users u ON ar.user_id = u.user_id AND u.status = 1 LEFT JOIN subjects s ON ar.subject_id = s.subject_id ORDER BY ar.role_type """ return await execute_query(sql) @staticmethod async def create(user_id: int, role_type: str, subject_id: int = None) -> int: sql = """ INSERT INTO admin_roles (user_id, role_type, subject_id) VALUES (%s, %s, %s) """ return await execute_insert(sql, (user_id, role_type, subject_id)) @staticmethod async def delete(user_id: int) -> bool: sql = "DELETE FROM admin_roles WHERE user_id = %s" result = await execute_update(sql, (user_id,)) return result > 0 @staticmethod async def update_role(user_id: int, role_type: str, subject_id: int = None) -> bool: sql = """ UPDATE admin_roles SET role_type = %s, subject_id = %s WHERE user_id = %s """ result = await execute_update(sql, (role_type, subject_id, user_id)) return result > 0