api/class_api.py

@@ -0,0 +1,78 @@
+from fastapi import APIRouter,Depends,HTTPException,Query
+from dao.class_dao import add_class_dao, update_class_dao, delete_class_dao, hard_delete_class_dao,class_no_exist,get_class_list_dao,get_class_by_id_dao
+from schemas.class_request import ClassCreate,ClassResponse,ClassUpdate
+from database import get_db
+from typing import Optional
+from sqlalchemy.orm import Session
+class_app = APIRouter()
+@class_app.post('',response_model=ClassResponse)
+def add_class(c:ClassCreate,db=Depends(get_db)):
+    if class_no_exist(c.class_id,db):
+        raise HTTPException(status_code=404,detail='该编号已经存在')
+    ok,result = add_class_dao(c.model_dump(),db)
+    if not ok:
+        raise HTTPException(status_code=500,detail='服务器繁忙或指定的老师不存在')
+    return ClassResponse(totals=1,data=result)
+@class_app.put('{id}')
+def update_class(id:int,c:ClassUpdate,db=Depends(get_db)):
+    d = c.model_dump()
+    r = update_class_dao(id=id ,update_data=d,db=db)
+    if not r:
+        raise HTTPException(status_code=500,detail='更新失败')
+    return{'code':200,'total':1,'detail':'更新成功'}
+'''
+@class_app.get('')
+def get_class(id:int,db=Depends(get_db),page:int=1,page_size:int=10):
+    rows = get_class_dao(id=id,db=db,page=page,page_size=page_size)
+    if rows:
+        return rows
+    raise HTTPException(status_code=404,detail='该班级不存在')
+'''
+@class_app.get('')
+def get_classes(
+        id: Optional[int] = Query(None, description="班级ID"),
+        class_id: Optional[str] = Query(None, description="班级编号"),
+        class_name: Optional[str] = Query(None, description="班级名称（支持模糊搜索）"),
+        ht_id: Optional[int] = Query(None, description="班主任ID"),
+        t_id: Optional[int] = Query(None, description="教师ID"),
+        page: int = Query(1, ge=1, description="页码"),
+        page_size: int = Query(10, ge=1, le=100, description="每页条数"),
+        db: Session = Depends(get_db)
+):
+    # 如果明确传了 id，当作单条查询处理
+    if id is not None:
+        single_class = get_class_by_id_dao(id=id, db=db)
+        if not single_class:
+            raise HTTPException(status_code=404, detail="该班级不存在")
+        return single_class
+
+    # 多条件列表查询
+    total, items = get_class_list_dao(
+        db=db,
+        class_no=class_id,
+        class_name=class_name,
+        ht_id=ht_id,
+        t_id=t_id,
+        page=page,
+        page_size=page_size
+    )
+
+    return {
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+        "items": items
+    }
+@class_app.delete('/{id}') #软删除
+def delete_class(id:int,db=Depends(get_db)):
+    rows = delete_class_dao(id=id,db=db)
+    if not rows:
+            raise HTTPException(status_code=404,detail='该班级不存在或已删除')
+    return{'code':200,'total':rows,'detail':'删除成功'}
+
+@class_app.delete('{id}/hard')
+def hard_delete_class(id:int,db=Depends(get_db)):
+    rows = hard_delete_class_dao(id=id,db=db)
+    if not rows:
+        raise HTTPException(status_code=404,detail='该班级不存在')
+    return{'code':200,'total':rows,'detail':'物理删除成功'}

dao/class_dao.py

@@ -0,0 +1,118 @@
+from pymysql import IntegrityError
+from models.class_model import Class,Teacher
+from sqlalchemy.orm import Session
+def add_class_dao(c,db):
+    try:
+        head_teacher = db.query(Teacher).filter(Teacher.id == c['ht_id']).first()
+        if head_teacher is None:
+           return False, '指定的班主任不存在'
+        class_teacher = db.query(Teacher).filter(Teacher.id == c['t_id']).first()
+        if class_teacher is None:
+           return False, '指定的授课老师不存在'
+        if db.query(Class).filter(
+                Class.class_id == c['class_id'],#如果请求体传进来的classid在表里找的到
+                Class.is_deleted==0 #且没有处于软删除状态
+        ).first():
+            return False, '班级编号已存在'
+        new_class = Class(**c)
+        db.add(new_class)
+        db.commit()
+        return True,new_class
+    except Exception as e:
+        db.rollback()
+        print('创建失败：',repr(e))
+        return False,str(e)
+def update_class_dao(id,update_data,db):
+    try:
+        rows = db.query(Class).filter(Class.id == id).update(update_data)
+    except Exception as e:
+        db.rollback()
+        print('更新失败：', repr(e))
+        return False, str(e)
+    else:
+        db.commit()
+        return rows
+'''
+def get_class_dao(id,db,page,page_size):
+        q = db.query(Class)
+        if id:
+            q = q.filter(Class.id == id,Class.is_deleted==0)
+        rows = q.offset((page-1)*page_size ).limit(page_size).all()
+        if rows:
+            return[{'class_id':i.id,'class_no':i.class_no,'class_name':i.class_name,'ht_id':i.ht_id,'t_id':i.t_id,'start_time':i.start_time}for i in rows]
+'''
+def get_class_list_dao(db: Session, page: int, page_size: int, **filters):
+    #软删除过滤
+    q = db.query(Class).filter(Class.is_deleted == 0)
+
+    # 精确匹配字段
+    if filters.get("class_no"):
+        q = q.filter(Class.class_id == filters["class_id"])
+    if filters.get("ht_id") is not None:
+        q = q.filter(Class.ht_id == filters["ht_id"])
+    if filters.get("t_id") is not None:
+        q = q.filter(Class.t_id == filters["t_id"])
+
+    # 模糊匹配字段（例如班级名称）
+    if filters.get("class_name"):
+        q = q.filter(Class.class_name.like(f"%{filters['class_name']}%"))
+
+    # 获取满足条件的总记录数（分页必备）
+    total = q.count()
+
+    # 分页查询
+    rows = q.offset((page - 1) * page_size).limit(page_size).all()
+
+    items = [ #如果匹配不到，则会返回空列表
+        {
+            'id': i.id,
+            'class_id': i.class_id,
+            'class_name': i.class_name,
+            'ht_id': i.ht_id,
+            't_id': i.t_id,
+            'start_time': i.start_time
+        }
+        for i in rows
+    ]
+    return total, items
+def get_class_by_id_dao(id: int, db: Session):
+    i = db.query(Class).filter(Class.id == id, Class.is_deleted == 0).first()
+    if not i:
+        return None
+    return {
+        'id': i.id,
+        'class_id': i.class_id,
+        'class_name': i.class_name,
+        'ht_id': i.ht_id,
+        't_id': i.t_id,
+        'start_time': i.start_time
+    }
+def delete_class_dao(id,db): #软删除
+    try:
+        rows = (db.query(Class)
+                .filter(Class.id == id,Class.is_deleted==0)#只匹配还没处于软删除状态的行
+                .update({Class.is_deleted:1},synchronize_session=False))#软删除就是把这个字段改成True
+    except Exception as e:
+        db.rollback()
+        print('删除失败：', repr(e))
+        return False, str(e)
+    else:
+        db.commit()
+        return rows
+def hard_delete_class_dao(id,db): #物理删除
+    try:
+        rows = (db.query(Class).filter(Class.id == id).delete(synchronize_session=False))
+        db.commit()
+        return rows
+    except IntegrityError:
+        db.rollback()
+        return False, '该班级被其他数据引用，无法彻底删除'
+    except Exception as e:
+        db.rollback()
+        print('物理删除失败', repr(e))
+        return False, str(e)
+def class_no_exist(class_id,db):
+    return (db.query(Class)
+            .filter(Class.class_id == class_id,Class.is_deleted==0)
+            .first() is not None
+            )

model/class_model.py

@@ -0,0 +1,35 @@
+from database import DATETIME,Base,Column,Integer,String,ForeignKey,Boolean,Enum as SQLEnum,DateTime,DATETIME
+from datetime import datetime
+from database import engine
+from enum import Enum
+class Class(Base):
+    '''
+    班级信息数据库类模型
+    '''
+    __tablename__ = "s_class"
+    id = Column(Integer,
+                primary_key=True,
+                autoincrement=True,
+                index=True,#为该列创建索引，加快查询速度
+                comment='班级主键id')
+    class_id = Column(String(32),
+                      nullable=False,
+                      index=True,
+                      comment='班级编号')
+    class_name = Column(String(32),
+                        nullable=False,
+                        index=True,
+                        comment='班级名字')
+    ht_id = Column(Integer,
+                   ForeignKey('s_teacher.id'),
+                            nullable=False,
+                            comment='班主任ID')
+    t_id = Column(Integer,
+                  ForeignKey('s_teacher.id'),
+                             nullable=False,
+                             comment='授课老师ID')
+    start_time = Column(DATETIME,default=datetime.now,comment='开课时间')
+    update_time = Column(DATETIME,default=datetime.now,onupdate=datetime.now)
+    is_deleted = Column(Integer,default=0,index=True)
+Base.metadata.create_all(engine)
+

schema/class_request.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel,Field,ConfigDict
+from datetime import datetime
+class ClassRequest(BaseModel): #定义更新班级信息的请求体
+    class_id: str = Field(...,)
+    class_name: str = Field(...,)
+    ht_id: int = Field(...)
+    t_id: int = Field(...)
+    start_time: datetime = Field(...)
+    model_config = ConfigDict(from_attributes=True)
+class ClassCreate(ClassRequest):
+    pass
+class ClassUpdate(ClassRequest):
+    pass
+class ClassResponse(BaseModel): #定义响应体
+    code:int = 200
+    detail:str ='OK'
+    totals:int = 0
+    data:ClassRequest | None = None #原来的写法str | dict | list | tuple 这几种类型里，
+                                    #一个都匹配不上路由里传的ORM对象result（Class类的实例），所以报错500
+
+
+