62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
# ===========================================
|
|
# 班级操行分管理系统 - 管理员角色模型
|
|
#
|
|
# 开发者: 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 |