60 lines
2.1 KiB
Python
60 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.*, c.class_name, s.subject_name
|
|
FROM admin_roles ar
|
|
LEFT JOIN classes c ON ar.class_id = c.class_id
|
|
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_class_id_by_user(user_id: int) -> Optional[int]:
|
|
sql = "SELECT class_id FROM admin_roles WHERE user_id = %s LIMIT 1"
|
|
result = await execute_one(sql, (user_id,))
|
|
return result["class_id"] if result else None
|
|
|
|
@staticmethod
|
|
async def get_by_class(class_id: int) -> List[Dict[str, Any]]:
|
|
sql = """
|
|
SELECT ar.*, u.real_name, u.username
|
|
FROM admin_roles ar
|
|
JOIN users u ON ar.user_id = u.user_id
|
|
WHERE ar.class_id = %s
|
|
ORDER BY ar.role_type
|
|
"""
|
|
return await execute_query(sql, (class_id,))
|
|
|
|
@staticmethod
|
|
async def create(user_id: int, role_type: str, class_id: int, subject_id: int = None) -> int:
|
|
sql = """
|
|
INSERT INTO admin_roles (user_id, role_type, class_id, subject_id)
|
|
VALUES (%s, %s, %s, %s)
|
|
"""
|
|
return await execute_insert(sql, (user_id, role_type, class_id, 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 |