成绩模块+课程模块
This commit is contained in:
@@ -0,0 +1,29 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from schema.course_request import CourseRequest, CourseResponse
|
||||
from database import get_db
|
||||
from dao.course_dao import (
|
||||
add_course_dao, get_all_courses_dao, get_course_by_id_dao
|
||||
)
|
||||
|
||||
course_api = APIRouter()
|
||||
|
||||
|
||||
@course_api.post('', response_model=CourseResponse, summary='添加课程')
|
||||
def add_course(course: CourseRequest, db=Depends(get_db)):
|
||||
d = course.model_dump()
|
||||
r = add_course_dao(o=d, db=db)
|
||||
if not r:
|
||||
raise HTTPException(status_code=500, detail='添加失败！')
|
||||
return CourseResponse(totals=1, data=d)
|
||||
|
||||
|
||||
@course_api.get('/{course_id}', summary='按编号查课程')
|
||||
def get_course(course_id: str, db=Depends(get_db)):
|
||||
c = get_course_by_id_dao(course_id=course_id, db=db)
|
||||
if not c:
|
||||
raise HTTPException(status_code=404, detail='课程不存在')
|
||||
return {
|
||||
'course_id': c.course_id,
|
||||
'course_name': c.course_name,
|
||||
'teacher_id': c.teacher_id,
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from schema.score_request import ScoreRequest, ScoreUpdateRequest, ScoreResponse
|
||||
from database import get_db
|
||||
from dao.score_dao import (
|
||||
calc_level, calc_pass,
|
||||
add_score_dao, update_score_dao, delete_score_dao, get_scores_dao
|
||||
)
|
||||
|
||||
score_api = APIRouter()
|
||||
|
||||
|
||||
# ========== 1. 添加成绩（按考核序次录入） ==========
|
||||
@score_api.post('', response_model=ScoreResponse, summary='添加成绩')
|
||||
def add_score(score: ScoreRequest, db=Depends(get_db)):
|
||||
d = score.model_dump()
|
||||
d['score_level'] = calc_level(d['score']) # ← 调 DAO 的工具函数
|
||||
d['is_pass'] = calc_pass(d['score'])
|
||||
|
||||
r = add_score_dao(o=d, db=db)
|
||||
if not r:
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
detail='添加失败！'
|
||||
)
|
||||
return ScoreResponse(totals=1, data=d)
|
||||
|
||||
|
||||
# ========== 2. 查询成绩（动态条件 + 分页） ==========
|
||||
@score_api.get('', summary='查询成绩（支持多条件+分页）')
|
||||
def get_scores(
|
||||
score_id: int = None,
|
||||
student_id: str = None,
|
||||
course_id: str = None,
|
||||
teacher_id: str = None,
|
||||
exam_order: int = None,
|
||||
page: int = 1,
|
||||
page_size: int = 10,
|
||||
db=Depends(get_db)
|
||||
):
|
||||
r = get_scores_dao(
|
||||
score_id=score_id, student_id=student_id, course_id=course_id,
|
||||
teacher_id=teacher_id, exam_order=exam_order,
|
||||
page=page, page_size=page_size, db=db
|
||||
)
|
||||
if r:
|
||||
return r
|
||||
raise HTTPException(status_code=404, detail='成绩不存在！')
|
||||
|
||||
|
||||
# ========== 3. 修改成绩 ==========
|
||||
@score_api.put('/{score_id}', summary='修改成绩')
|
||||
def update_score(score_id: int, score: ScoreUpdateRequest, db=Depends(get_db)):
|
||||
d = score.model_dump(exclude_unset=True)
|
||||
if 'score' in d:
|
||||
d['score_level'] = calc_level(d['score'])
|
||||
d['is_pass'] = calc_pass(d['score'])
|
||||
|
||||
r = update_score_dao(score_id=score_id, update_data=d, db=db)
|
||||
if not r:
|
||||
raise HTTPException(status_code=500, detail='没有更新！')
|
||||
return {'code': 200, 'totals': r, 'detail': '更新成功'}
|
||||
|
||||
|
||||
# ========== 4. 删除成绩（逻辑删除） ==========
|
||||
@score_api.delete('/{score_id}', summary='删除成绩(逻辑删除)')
|
||||
def delete_score(score_id: int, db=Depends(get_db)):
|
||||
r = delete_score_dao(score_id=score_id, db=db)
|
||||
if not r:
|
||||
raise HTTPException(status_code=500, detail='删除异常，稍后操作！')
|
||||
return {'code': 200, 'totals': r, 'detail': '删除成功'}
|
||||
@@ -0,0 +1,24 @@
|
||||
from model.course_model import Course
|
||||
|
||||
|
||||
def add_course_dao(o, db):
|
||||
try:
|
||||
c = Course(**o)
|
||||
db.add(c)
|
||||
except:
|
||||
db.rollback()
|
||||
return False
|
||||
else:
|
||||
db.commit()
|
||||
return True
|
||||
|
||||
|
||||
def get_all_courses_dao(db):
|
||||
return db.query(Course).filter(Course.is_deleted == 0).all()
|
||||
|
||||
|
||||
def get_course_by_id_dao(course_id, db):
|
||||
return db.query(Course).filter(
|
||||
Course.course_id == course_id,
|
||||
Course.is_deleted == 0
|
||||
).first()
|
||||
@@ -0,0 +1,98 @@
|
||||
from model.score_model import ScoreRecord
|
||||
|
||||
|
||||
# ========== 工具函数：算等级 ==========
|
||||
def calc_level(score):
|
||||
if score is None: return None
|
||||
if score >= 95: return 'A+'
|
||||
if score >= 90: return 'A'
|
||||
if score >= 80: return 'B'
|
||||
if score >= 70: return 'C'
|
||||
if score >= 60: return 'D'
|
||||
return 'E'
|
||||
|
||||
|
||||
# ========== 工具函数：算及格 ==========
|
||||
def calc_pass(score):
|
||||
if score is None: return None
|
||||
return 1 if score >= 60 else 0
|
||||
|
||||
|
||||
# ========== 1. 添加成绩 ==========
|
||||
def add_score_dao(o, db):
|
||||
try:
|
||||
# 【业务校验】该学生该课程该序次是否已存在
|
||||
existing = db.query(ScoreRecord).filter(
|
||||
ScoreRecord.student_id == o['student_id'],
|
||||
ScoreRecord.course_id == o['course_id'],
|
||||
ScoreRecord.exam_order == o['exam_order'],
|
||||
ScoreRecord.is_deleted == 0
|
||||
).all()
|
||||
|
||||
if existing:
|
||||
raise ValueError('该学生该课程该序次成绩已存在')
|
||||
|
||||
o1 = ScoreRecord(**o)
|
||||
db.add(o1)
|
||||
except:
|
||||
db.rollback()
|
||||
return False
|
||||
else:
|
||||
db.commit()
|
||||
return True
|
||||
|
||||
|
||||
# ========== 2. 修改成绩 ==========
|
||||
def update_score_dao(score_id, update_data, db):
|
||||
try:
|
||||
rows = db.query(ScoreRecord).filter(
|
||||
ScoreRecord.score_id == score_id,
|
||||
ScoreRecord.is_deleted == 0
|
||||
).update(update_data)
|
||||
except:
|
||||
db.rollback()
|
||||
return False
|
||||
else:
|
||||
db.commit()
|
||||
return rows
|
||||
|
||||
|
||||
# ========== 3. 查询成绩（动态条件 + 分页） ==========
|
||||
def get_scores_dao(score_id, student_id, course_id, teacher_id,
|
||||
exam_order, page, page_size, db):
|
||||
q = db.query(ScoreRecord).filter(ScoreRecord.is_deleted == 0)
|
||||
|
||||
if score_id:
|
||||
q = q.filter(ScoreRecord.score_id == score_id)
|
||||
if student_id:
|
||||
q = q.filter(ScoreRecord.student_id == student_id)
|
||||
if course_id:
|
||||
q = q.filter(ScoreRecord.course_id == course_id)
|
||||
if teacher_id:
|
||||
q = q.filter(ScoreRecord.teacher_id == teacher_id)
|
||||
if exam_order:
|
||||
q = q.filter(ScoreRecord.exam_order == exam_order)
|
||||
|
||||
r = q.order_by(ScoreRecord.exam_order.asc()) \
|
||||
.offset((page - 1) * page_size) \
|
||||
.limit(page_size).all()
|
||||
|
||||
if r:
|
||||
return [item.to_dict() for item in r]
|
||||
return []
|
||||
|
||||
|
||||
# ========== 4. 删除成绩 ==========
|
||||
def delete_score_dao(score_id, db):
|
||||
try:
|
||||
rows = db.query(ScoreRecord).filter(
|
||||
ScoreRecord.score_id == score_id,
|
||||
ScoreRecord.is_deleted == 0
|
||||
).update({'is_deleted': 1})
|
||||
except:
|
||||
db.rollback()
|
||||
rows = 0
|
||||
else:
|
||||
db.commit()
|
||||
finally:
|
||||
return rows
|
||||
@@ -7,6 +7,8 @@ from api.student_info_api import student_info_api
|
||||
from api.region_api import region_api
|
||||
from api.teacher_api import TeacherAPI
|
||||
from api.teacher_course_api import TeacherCourseAPI
|
||||
from api.course_api import course_api
|
||||
from api.score_api import score_api
|
||||
from starlette.middleware.cors import CORSMiddleware
|
||||
|
||||
|
||||
@@ -31,6 +33,8 @@ app.include_router(TeacherAPI,tags=['老师信息'])
|
||||
|
||||
app.include_router(TeacherCourseAPI,tags=['老师课程信息'])
|
||||
|
||||
app.include_router(score_api,tags=['成绩信息'],prefix='/score')
|
||||
app.include_router(course_api,tags=['课程信息'],prefix='/course')
|
||||
app.include_router(employment_router)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
from database import DATETIME, Base, Column, Integer, String
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class Course(Base):
|
||||
__tablename__ = 'student_course'
|
||||
|
||||
course_id = Column(String(20), primary_key=True, comment='课程编号')
|
||||
course_name = Column(String(50), nullable=False, comment='课程名称')
|
||||
teacher_id = Column(String(20), nullable=True, comment='授课教师编号')
|
||||
is_deleted = Column(Integer, nullable=False, default=0, comment='0未删1已删')
|
||||
create_time = Column(DATETIME, default=datetime.now, nullable=False)
|
||||
update_time = Column(DATETIME, default=datetime.now, onupdate=datetime.now, nullable=False)
|
||||
@@ -0,0 +1,79 @@
|
||||
from database import DATETIME, Base, Column, Integer, String, Numeric, Date
|
||||
from datetime import datetime
|
||||
from sqlalchemy import ForeignKey
|
||||
|
||||
|
||||
# 考试类型映射
|
||||
EXAM_TYPE_MAP = {
|
||||
1: '期中',
|
||||
2: '期末',
|
||||
3: '月考',
|
||||
4: '模拟',
|
||||
}
|
||||
|
||||
|
||||
class ScoreRecord(Base):
|
||||
__tablename__ = 'student_score'
|
||||
|
||||
score_id = Column(Integer, primary_key=True, autoincrement=True,
|
||||
comment='成绩记录自增主键')
|
||||
student_id = Column(
|
||||
String(20),
|
||||
# ForeignKey("student_info.student_id", ondelete="RESTRICT"), # 等组员表定稿再加
|
||||
nullable=False,
|
||||
comment='学号'
|
||||
)
|
||||
|
||||
exam_order = Column(
|
||||
Integer,
|
||||
nullable=False,
|
||||
default=1,
|
||||
comment='考核序次（第几次）'
|
||||
)
|
||||
|
||||
course_id = Column(
|
||||
String(20),
|
||||
# ForeignKey("student_course.course_id", ondelete="RESTRICT"),
|
||||
nullable=False,
|
||||
comment='课程编号'
|
||||
)
|
||||
teacher_id = Column(
|
||||
String(20),
|
||||
# ForeignKey("teacher_info.teacher_id", ondelete="RESTRICT"), # 等组员表定稿再加
|
||||
nullable=False,comment='教师编号'
|
||||
)
|
||||
exam_type = Column(Integer, nullable=False, comment='1期中 2期末 3月考 4模拟')
|
||||
|
||||
score = Column(Numeric(5, 2), nullable=True, comment='分数0~100，缺考NULL')
|
||||
|
||||
score_level = Column(String(2), nullable=True, comment='等级A+~E')
|
||||
|
||||
is_pass = Column(Integer, nullable=True, default=0, comment='1及格 0不及格')
|
||||
|
||||
exam_date = Column(Date, nullable=False, comment='考试日期')
|
||||
|
||||
remark = Column(String(200), nullable=True, comment='缺考,缓考')
|
||||
|
||||
is_deleted = Column(Integer, nullable=False, default=0, comment='0未删 1已删')
|
||||
|
||||
create_time = Column(DATETIME, default=datetime.now, nullable=False)
|
||||
|
||||
update_time = Column(DATETIME, default=datetime.now,
|
||||
onupdate=datetime.now, nullable=False)
|
||||
|
||||
# ========== 实例方法：转字典 ==========
|
||||
def to_dict(self):
|
||||
return {
|
||||
'score_id': self.score_id,
|
||||
'student_id': self.student_id,
|
||||
'course_id': self.course_id,
|
||||
'teacher_id': self.teacher_id,
|
||||
'exam_order': self.exam_order,
|
||||
'exam_type': self.exam_type,
|
||||
'exam_type_name': EXAM_TYPE_MAP.get(self.exam_type, '未知'),
|
||||
'score': float(self.score) if self.score is not None else None,
|
||||
'score_level': self.score_level,
|
||||
'is_pass': self.is_pass,
|
||||
'exam_date': str(self.exam_date),
|
||||
'remark': self.remark,
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
from pydantic import BaseModel, Field
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class CourseRequest(BaseModel):
|
||||
course_id: str = Field(..., description='课程编号')
|
||||
course_name: str = Field(..., description='课程名称')
|
||||
teacher_id: Optional[str] = Field(None, description='授课教师编号')
|
||||
|
||||
|
||||
class CourseResponse(BaseModel):
|
||||
code: int = 200
|
||||
detail: str = 'OK'
|
||||
totals: int = 0
|
||||
data: str | dict | tuple | list
|
||||
@@ -0,0 +1,29 @@
|
||||
from pydantic import BaseModel, Field
|
||||
from datetime import date
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class ScoreRequest(BaseModel):
|
||||
student_id: str = Field(...,min_length=1, max_length=20,description="学号")
|
||||
course_id: str = Field(..., min_length=1, max_length=20,description="课程编号")
|
||||
teacher_id: str = Field(..., min_length=1, max_length=20,description="教师编号")
|
||||
exam_order: int = Field(1,ge=1, description="考核序次(第几次考核）")
|
||||
exam_type: int = Field(...,ge=1, le=4,description="考试类型：1=期中，2=期末，3=月考，4=模拟")
|
||||
score: Optional[float] = Field(None,ge=0, le=100,description="分数 0~100，缺考填 null")
|
||||
exam_date: date = Field(..., description="考试日期（YYYY-MM-DD）")
|
||||
remark: Optional[str] = Field(None, description="备注，例如：缺考、缓考")
|
||||
|
||||
|
||||
class ScoreUpdateRequest(BaseModel):
|
||||
score: Optional[float] = Field(
|
||||
None,
|
||||
ge=0, le=100,
|
||||
description="新分数 0~100"
|
||||
)
|
||||
remark: Optional[str] = Field(None, description="新备注")
|
||||
|
||||
class ScoreResponse(BaseModel):
|
||||
code: int = 200
|
||||
detail: str = 'OK'
|
||||
totals: int = 0
|
||||
data: str | dict | tuple | list
|
||||
Reference in New Issue
Block a user