89 lines
3.3 KiB
Python
89 lines
3.3 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 SubjectModel:
|
||
"""科目数据模型"""
|
||
|
||
@staticmethod
|
||
async def get_all(is_active: bool = None) -> List[Dict[str, Any]]:
|
||
if is_active is not None:
|
||
sql = "SELECT * FROM subjects WHERE is_active = %s ORDER BY sort_order, subject_id"
|
||
return await execute_query(sql, (1 if is_active else 0,))
|
||
else:
|
||
sql = "SELECT * FROM subjects ORDER BY sort_order, subject_id"
|
||
return await execute_query(sql)
|
||
|
||
@staticmethod
|
||
async def get_by_id(subject_id: int) -> Optional[Dict[str, Any]]:
|
||
sql = "SELECT * FROM subjects WHERE subject_id = %s"
|
||
return await execute_one(sql, (subject_id,))
|
||
|
||
@staticmethod
|
||
async def get_by_name(subject_name: str) -> Optional[Dict[str, Any]]:
|
||
sql = "SELECT * FROM subjects WHERE subject_name = %s"
|
||
return await execute_one(sql, (subject_name,))
|
||
|
||
@staticmethod
|
||
async def create(subject_name: str, subject_code: str = None, sort_order: int = 0) -> int:
|
||
sql = """
|
||
INSERT INTO subjects (subject_name, subject_code, sort_order)
|
||
VALUES (%s, %s, %s)
|
||
"""
|
||
return await execute_insert(sql, (subject_name, subject_code, sort_order))
|
||
|
||
@staticmethod
|
||
async def update(subject_id: int, **kwargs) -> bool:
|
||
updates = []
|
||
params = []
|
||
|
||
if "subject_name" in kwargs:
|
||
updates.append("subject_name = %s")
|
||
params.append(kwargs["subject_name"])
|
||
if "subject_code" in kwargs:
|
||
updates.append("subject_code = %s")
|
||
params.append(kwargs["subject_code"])
|
||
if "is_active" in kwargs:
|
||
updates.append("is_active = %s")
|
||
params.append(1 if kwargs["is_active"] else 0)
|
||
if "sort_order" in kwargs:
|
||
updates.append("sort_order = %s")
|
||
params.append(kwargs["sort_order"])
|
||
|
||
if not updates:
|
||
return True
|
||
|
||
params.append(subject_id)
|
||
sql = f"UPDATE subjects SET {', '.join(updates)} WHERE subject_id = %s"
|
||
result = await execute_update(sql, tuple(params))
|
||
return result > 0
|
||
|
||
@staticmethod
|
||
async def has_related_data(subject_id: int) -> bool:
|
||
"""检查科目是否有关联数据（assignments表）"""
|
||
sql = "SELECT COUNT(*) as cnt FROM assignments WHERE subject_id = %s"
|
||
result = await execute_one(sql, (subject_id,))
|
||
return result['cnt'] > 0
|
||
|
||
@staticmethod
|
||
async def delete(subject_id: int) -> bool:
|
||
sql = "UPDATE subjects SET is_active = 0 WHERE subject_id = %s"
|
||
result = await execute_update(sql, (subject_id,))
|
||
return result > 0
|
||
|
||
@staticmethod
|
||
async def activate(subject_id: int) -> bool:
|
||
sql = "UPDATE subjects SET is_active = 1 WHERE subject_id = %s"
|
||
result = await execute_update(sql, (subject_id,))
|
||
return result > 0 |