Merge pull request '成绩模块+课程模块' (#5) from cjg into master

Reviewed-on: #5

api/course_api.py

@@ -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,
+    }

api/score_api.py

@@ -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': '删除成功'}

api/teacher_api.py

@@ -0,0 +1,41 @@
+from fastapi import APIRouter,Depends,HTTPException
+from schema.teacher_schema import TeacherRequest,TeacherResponse,TeacherUpdateRequest
+from database import get_db
+from dao.teacher_dao import add_teacher_dao,update_teacher_dao,get_teacher_dao,delete_teacher_dao
+
+TeacherAPI = APIRouter()
+
+@TeacherAPI.post('/teachers',response_model=TeacherResponse,summary='新增老师')
+def add_teachers(teacher:TeacherRequest,db=Depends(get_db)):
+    d = teacher.model_dump()
+    r = add_teacher_dao(t=d,db=db)
+    if not r:
+        raise HTTPException(status_code=400,detail='教师编号已存在或新增失败！')
+    return teacher
+
+
+@TeacherAPI.put('/teachers/{teacher_id}',summary='更新老师')
+def update_teachers(teacher_id:str,teacher:TeacherUpdateRequest,db=Depends(get_db)):
+    d = teacher.model_dump(exclude_unset=True)
+    r = update_teacher_dao(teacher_id=teacher_id,data=d,db=db) #返回更新的行数
+    if not r:
+        raise HTTPException(status_code=404,detail='老师不存在或未更新！')
+    return {'code':200,'totals':r,'detail':'更新成功'}
+
+
+@TeacherAPI.get('/teachers',response_model=list[TeacherResponse],summary='查询老师')
+def get_teachers(teacher_id:str|None=None,
+                 teacher_name:str|None=None,
+                 db=Depends(get_db)):
+    r = get_teacher_dao(teacher_id=teacher_id,teacher_name=teacher_name,db=db)
+    if not r:
+        raise HTTPException(status_code=404,detail='该老师不存在！')
+    return r
+
+
+@TeacherAPI.delete('/teachers/{teacher_id}',summary='删除老师')
+def delete_teachers(teacher_id:str|None = None,db=Depends(get_db)):
+    r = delete_teacher_dao(teacher_id,db)
+    if not r:
+        raise HTTPException(status_code=404,detail='老师不存在或删除失败！')
+    return {'code':200,'totals':r,'detail':'删除成功！'}

api/teacher_course_api.py

@@ -0,0 +1,25 @@
+from fastapi import APIRouter,Depends,HTTPException
+from schema.teacher_schema import TeacherCourseRequest,TeacherCourseResponse
+from database import get_db
+from dao.teacher_course_dao import add_teacher_course_dao,get_teacher_course_dao
+
+
+TeacherCourseAPI = APIRouter()
+
+
+@TeacherCourseAPI.post('/teacher_courses',response_model=TeacherCourseResponse,summary='新增老师课程')
+def add_teacher_courses(teachercourse:TeacherCourseRequest,db=Depends(get_db)):
+    d = teachercourse.model_dump()
+    r = add_teacher_course_dao(t=d,db=db)
+    if not r:
+        raise HTTPException(status_code=400,detail='新增课程失败！')
+    return d
+
+
+@TeacherCourseAPI.get('/teacher_courses',response_model=list[TeacherCourseResponse],summary='查询老师课程')
+def get_teacher_courses(teacher_id:str,
+                 db=Depends(get_db)):
+    r = get_teacher_course_dao(teacher_id=teacher_id,db=db)
+    if not r:
+        raise HTTPException(status_code=404,detail='该老师不存在！')
+    return r

dao/course_dao.py

@@ -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()

dao/score_dao.py

@@ -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

dao/teacher_course_dao.py

@@ -0,0 +1,24 @@
+from model.teacher_model import TeacherCourse
+
+
+def add_teacher_course_dao(t,db):
+    try:   #判断是都已存在
+        r = db.query(TeacherCourse).filter(TeacherCourse.teacher_id==t['teacher_id']).first()
+        if r:
+            return None   #已存在，交给上层判断
+        obj = TeacherCourse(**t)
+        db.add(obj)
+        db.commit()
+        db.refresh(obj)  #拿回自增id，create_time等
+        return obj
+    except Exception:
+        db.rollback()
+        return False
+
+
+def get_teacher_course_dao(teacher_id=None,db=None):
+    q = db.query(TeacherCourse)
+    if teacher_id:
+        q = q.filter(TeacherCourse.teacher_id==teacher_id)
+    return q.all()   #返回ORM对象列表
+

dao/teacher_dao.py

@@ -0,0 +1,53 @@
+from model.teacher_model import Teacher
+
+
+def add_teacher_dao(t,db):
+    try:   #判断是都已存在
+        r = db.query(Teacher).filter(Teacher.teacher_id==t['teacher_id']).first()
+        if r:
+            return None   #已存在，交给上层判断
+        obj = Teacher(**t)
+        db.add(obj)
+
+        db.commit()
+        db.refresh(obj)  #拿回自增id，create_time等
+        return obj
+    except Exception:
+        db.rollback()
+        return False
+
+def update_teacher_dao(teacher_id,data,db):   #data:新增的内容
+    try:   #数据更新有返回值，返回的是影响的行数
+        rows = db.query(Teacher).filter(Teacher.teacher_id == teacher_id).update(data)
+        db.commit()
+        return rows
+    except Exception:
+        db.rollback()
+        return False
+
+
+
+
+def get_teacher_dao(teacher_id=None,teacher_name=None,db=None):
+    q = db.query(Teacher)
+    if teacher_id:
+        q = q.filter(Teacher.teacher_id==teacher_id)
+    if teacher_name:
+        q = q.filter(Teacher.teacher_name.like (f'%{teacher_name}%'))
+    return q.all()   #返回ORM对象列表
+
+
+
+def delete_teacher_dao(teacher_id,db):
+    try:    #删除有返回值，删除的行数
+        rows = (db.query(Teacher)
+             .filter(Teacher.teacher_id == teacher_id)
+             .delete())
+        db.commit()
+        return rows
+    except Exception:
+        db.rollback()
+        return 0
+
+
+

main.py

@@ -5,6 +5,10 @@ from api.calss import class_api
 from api.employment_api import app as employment_router
 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
 
 
@@ -24,6 +28,13 @@ app.include_router(student_info_api,tags=['学生信息'])
 
 app.include_router(region_api,tags=['地区信息'])
 
+
+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)
 
 

model/course_model.py

@@ -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)

model/score_model.py

@@ -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,
+        }

model/teacher_model.py

@@ -0,0 +1,40 @@
+from database import DATETIME,Base,engine
+from datetime import datetime
+from sqlalchemy import Column,String,Date,Integer
+
+
+
+class Teacher(Base):
+    __tablename__ = 'teacher_info'
+    id = Column(Integer,primary_key=True,autoincrement=True,comment='自增主键id')
+    teacher_id = Column(String(20),nullable=False,unique=True,comment='教师编号')
+    teacher_name = Column(String(50),nullable=False,comment='老师姓名')
+    gender = Column(Integer,nullable=False,comment='性别:0=女,1=男')
+    birthday = Column(Date,nullable=True,comment='出生年月')
+    native_region_code = Column(String(12),nullable=True,comment='籍贯地区编码')
+    email = Column(String(100),nullable=True,comment='邮箱')
+    phone = Column(String(11),nullable=False,comment='手机号')
+    graduation_school = Column(String(20),nullable=True,comment='毕业院校')
+    is_deleted = Column(Integer,nullable=False,default=0,comment='逻辑删除：0=未删除，1=已删除')
+    create_time = Column(DATETIME,nullable=False,default=datetime.now,comment='创建时间')
+    update_time = Column(DATETIME,default=datetime.now,onupdate=datetime.now,comment='更新时间')
+
+
+
+class TeacherCourse(Base):
+    __tablename__ = 'teacher_course'
+    id = Column(Integer,primary_key=True,autoincrement=True,comment='记录ID')
+    teacher_id = Column(String(20),nullable=False,comment='教师编号')
+    course_id = Column(String(20),nullable=False,comment='任课科目编号')
+    is_deleted = Column(Integer,nullable=False,default=0,comment='逻辑删除：0=未删除，1=已删除')
+    create_time = Column(DATETIME,nullable=False,default=datetime.now,comment='创建时间')
+    update_time = Column(DATETIME,nullable=False,default=datetime.now,onupdate=datetime.now,comment='更新时间')
+
+
+
+if __name__ == "__main__":
+    Base.metadata.create_all(engine)
+    print("数据表创建完成！")
+
+
+

schema/course_request.py

@@ -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

schema/score_request.py

@@ -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

schema/teacher_schema.py

@@ -0,0 +1,45 @@
+from pydantic import BaseModel
+from datetime import date
+
+from typing import Literal
+
+class TeacherRequest(BaseModel):        #Teacher请求体模型
+    teacher_id: str
+    teacher_name: str
+    gender: Literal[1,0]
+    birthday: date|None = None
+    native_region_code: str|None = None
+    email: str|None = None
+    phone: str
+    graduation_school: str|None = None
+
+class TeacherUpdateRequest(BaseModel):
+    teacher_name: str|None = None
+    gender: Literal[1,0]|None = None
+    birthday: date | None = None
+    native_region_code: str | None = None
+    email: str | None = None
+    phone: str
+    graduation_school: str | None = None
+
+
+class TeacherResponse(BaseModel):         #Teacher响应体模型
+    teacher_id: str
+    teacher_name: str
+    gender: Literal[1,0]
+    birthday: date|None = None
+    native_region_code: str|None = None
+    email: str|None = None
+    phone: str
+    graduation_school: str|None = None
+
+class TeacherCourseRequest(BaseModel):       #TeacherCourse请求体模型
+    teacher_id: str|None = None
+    course_id: str|None = None
+
+class TeacherCourseResponse(BaseModel):    #老师带班信息及课程表响应体模型
+    teacher_id: str|None = None
+    course_id: str|None = None
+    model_config = {"from_attributes": True}     #可以直接读取ORM模型对象，不用手动转字典
+
+