998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
138 lines
4.2 KiB
Python
138 lines
4.2 KiB
Python
"""
|
|
课程考试设置服务
|
|
"""
|
|
from typing import Optional
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.logger import get_logger
|
|
from app.models.course_exam_settings import CourseExamSettings
|
|
from app.schemas.course import CourseExamSettingsCreate, CourseExamSettingsUpdate
|
|
from app.services.base_service import BaseService
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class CourseExamService(BaseService[CourseExamSettings]):
|
|
"""课程考试设置服务"""
|
|
|
|
def __init__(self):
|
|
super().__init__(CourseExamSettings)
|
|
|
|
async def get_by_course_id(self, db: AsyncSession, course_id: int) -> Optional[CourseExamSettings]:
|
|
"""
|
|
根据课程ID获取考试设置
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
course_id: 课程ID
|
|
|
|
Returns:
|
|
考试设置实例或None
|
|
"""
|
|
stmt = select(CourseExamSettings).where(
|
|
CourseExamSettings.course_id == course_id,
|
|
CourseExamSettings.is_deleted == False
|
|
)
|
|
result = await db.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def create_or_update(
|
|
self,
|
|
db: AsyncSession,
|
|
course_id: int,
|
|
settings_in: CourseExamSettingsCreate,
|
|
user_id: int
|
|
) -> CourseExamSettings:
|
|
"""
|
|
创建或更新课程考试设置
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
course_id: 课程ID
|
|
settings_in: 考试设置数据
|
|
user_id: 操作用户ID
|
|
|
|
Returns:
|
|
考试设置实例
|
|
"""
|
|
# 检查是否已存在设置
|
|
existing_settings = await self.get_by_course_id(db, course_id)
|
|
|
|
if existing_settings:
|
|
# 更新现有设置
|
|
update_data = settings_in.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(existing_settings, field, value)
|
|
existing_settings.updated_by = user_id
|
|
|
|
await db.commit()
|
|
await db.refresh(existing_settings)
|
|
|
|
logger.info(f"更新课程考试设置成功", course_id=course_id, user_id=user_id)
|
|
return existing_settings
|
|
else:
|
|
# 创建新设置
|
|
create_data = settings_in.model_dump()
|
|
create_data.update({
|
|
"course_id": course_id,
|
|
"created_by": user_id,
|
|
"updated_by": user_id
|
|
})
|
|
|
|
new_settings = CourseExamSettings(**create_data)
|
|
db.add(new_settings)
|
|
|
|
await db.commit()
|
|
await db.refresh(new_settings)
|
|
|
|
logger.info(f"创建课程考试设置成功", course_id=course_id, user_id=user_id)
|
|
return new_settings
|
|
|
|
async def update(
|
|
self,
|
|
db: AsyncSession,
|
|
course_id: int,
|
|
settings_in: CourseExamSettingsUpdate,
|
|
user_id: int
|
|
) -> CourseExamSettings:
|
|
"""
|
|
更新课程考试设置
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
course_id: 课程ID
|
|
settings_in: 更新的考试设置数据
|
|
user_id: 操作用户ID
|
|
|
|
Returns:
|
|
更新后的考试设置实例
|
|
"""
|
|
# 获取现有设置
|
|
settings = await self.get_by_course_id(db, course_id)
|
|
if not settings:
|
|
# 如果不存在，创建新的
|
|
create_data = settings_in.model_dump(exclude_unset=True)
|
|
return await self.create_or_update(
|
|
db,
|
|
course_id=course_id,
|
|
settings_in=CourseExamSettingsCreate(**create_data),
|
|
user_id=user_id
|
|
)
|
|
|
|
# 更新设置
|
|
update_data = settings_in.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(settings, field, value)
|
|
settings.updated_by = user_id
|
|
|
|
await db.commit()
|
|
await db.refresh(settings)
|
|
|
|
logger.info(f"更新课程考试设置成功", course_id=course_id, user_id=user_id)
|
|
return settings
|
|
|
|
|
|
# 创建服务实例
|
|
course_exam_service = CourseExamService()
|