班级模块
This commit is contained in:
+164
@@ -0,0 +1,164 @@
|
||||
import uuid
|
||||
from fastapi import APIRouter, Depends, HTTPException,Query
|
||||
from dao.class_dao import update_class_dao, add_class_dao, get_class_dao
|
||||
from database import get_db
|
||||
from model.class_model import StudentClassIntermediate,Student_info
|
||||
from schema.class_request import ClassRequest, ClassResponse, TransferRequest
|
||||
from dao.class_dao import Student_delect_Dao
|
||||
from datetime import datetime
|
||||
from model.class_model import Class_info
|
||||
from sqlalchemy import select, func
|
||||
|
||||
|
||||
|
||||
class_api = APIRouter()
|
||||
|
||||
|
||||
@class_api.post('',response_model=ClassResponse,summary='班级信息新增接口')
|
||||
def add_orders(cs:ClassRequest,db=Depends(get_db)):
|
||||
d = cs.model_dump()
|
||||
r = add_class_dao( o=d,db=db )
|
||||
if not r:
|
||||
raise HTTPException(status_code=500,detail='服务器繁忙或班级不存在！')
|
||||
return ClassResponse(totals=1,data=d)
|
||||
|
||||
|
||||
# @class_api.delete('/{class_name}/')
|
||||
# def delete_orders(class_name:str,db=Depends(get_db)):
|
||||
# rows= delect_class_dao( class_name=class_name,db=db )
|
||||
# if rows:
|
||||
# return {'code': 200, 'totals': rows, 'detail': '删除成功'}
|
||||
# raise HTTPException(status_code=500,detail='删除异常，稍后操作！')
|
||||
|
||||
|
||||
|
||||
@class_api.delete("/class_id",summary='班级信息删除接口')
|
||||
async def delete_student(class_id: str, db= Depends(get_db)):
|
||||
target_student = Student_delect_Dao.soft_delete_student(db, class_id)
|
||||
if not target_student:
|
||||
raise HTTPException(status_code=404, detail="班级数据不存在或已删除")
|
||||
# 直接更新软删除标记为1，完成逻辑删除
|
||||
target_student.is_deleted = 1
|
||||
r=datetime.now().isoformat()
|
||||
db.commit()
|
||||
return ClassResponse(totals=1,data=r)
|
||||
|
||||
|
||||
|
||||
|
||||
@class_api.get('',summary='班级信息查询接口')
|
||||
def get_orders(class_name:str,db=Depends(get_db)):
|
||||
r=get_class_dao(class_name,db)
|
||||
if r:
|
||||
return [{"class_id": i.class_id, 'class_name': i.class_name, "class_code": i.class_code, "tags": i.tags,
|
||||
"status":i.status,'create_time':i.create_time,"update_time": i.update_time} for i in r]
|
||||
raise HTTPException(status_code=404,detail='班级不存在！')
|
||||
|
||||
@class_api.put('/{class_id}',summary='班级信息更新接口')
|
||||
def get_orders(order:ClassRequest,class_id:str,db=Depends(get_db)):
|
||||
d = order.model_dump(exclude_unset=True)
|
||||
r = update_class_dao( class_id=class_id,update_data=d,db=db )
|
||||
if not r:
|
||||
raise HTTPException(status_code=500, detail='没有更新！')
|
||||
return {'code':200,'totals':r,'detail':'更新成功'}
|
||||
|
||||
|
||||
@class_api.get("/classes",summary='班级学生信息分页查询接口')
|
||||
def get_all_classes_with_student_count(
|
||||
page: int = Query(1, ge=1), #page：页码，默认值为1
|
||||
size: int = Query(10, ge=1, le=100), #size：单页返回数据条数，默认值10，约束范围1~100，
|
||||
db = Depends(get_db) #db：通过依赖注入的方式获取SQLAlchemy的数据库会话实例
|
||||
):
|
||||
total = db.scalar(select(func.count(Class_info.class_id)).where(Class_info.is_deleted == 0)) #班级表中所有符合条件的主键数量
|
||||
offset = (page - 1) * size #计算分页偏移量
|
||||
|
||||
class_list = db.execute(
|
||||
select(
|
||||
Class_info.class_name,
|
||||
# 直接统计当前班级的学生数量
|
||||
(
|
||||
select(func.count(StudentClassIntermediate.student_id))
|
||||
.where(
|
||||
StudentClassIntermediate.class_id == Class_info.class_id,
|
||||
StudentClassIntermediate.is_deleted == 0
|
||||
)
|
||||
).label("student_count")
|
||||
)
|
||||
.where(Class_info.is_deleted == 0)
|
||||
.order_by(Class_info.class_name)
|
||||
.offset(offset).limit(size)
|
||||
).mappings().all()
|
||||
|
||||
return {
|
||||
"total_classes": total,
|
||||
"page": page,
|
||||
"size": size,
|
||||
"class_list": class_list
|
||||
}
|
||||
|
||||
|
||||
|
||||
# 同步转班接口
|
||||
@class_api.patch("/students/{student_id}/transfer",summary='学生转班信息查询接口')
|
||||
def transfer_student(
|
||||
student_id: str,
|
||||
request: TransferRequest,
|
||||
db=Depends(get_db)
|
||||
):
|
||||
target_class_name = request.target_class_name
|
||||
|
||||
# 1. 【关键修改】根据班级名称查找班级ID
|
||||
# 注意：这里假设班级名称是唯一的。如果名称不唯一，建议前端传ID或后端增加歧义处理
|
||||
target_class_res = db.execute(
|
||||
select(Class_info).where(
|
||||
Class_info.class_name == target_class_name,
|
||||
Class_info.is_deleted == 0
|
||||
)
|
||||
)
|
||||
target_class = target_class_res.scalar_one_or_none()
|
||||
|
||||
if not target_class:
|
||||
raise HTTPException(status_code=404, detail=f"未找到名为 '{target_class_name}' 的班级")
|
||||
|
||||
# 获取目标班级的ID，用于后续关联
|
||||
target_class_id = target_class.class_id
|
||||
# 2. 查询学生当前所在的有效班级关联
|
||||
current_link_res = db.execute(
|
||||
select(StudentClassIntermediate).where(
|
||||
StudentClassIntermediate.student_id == student_id,
|
||||
StudentClassIntermediate.is_deleted == 0
|
||||
)
|
||||
)
|
||||
current_link = current_link_res.scalar_one_or_none()
|
||||
|
||||
now = datetime.now()
|
||||
|
||||
# 3. 逻辑处理：如果学生已经在目标班级，直接返回
|
||||
if current_link and current_link.class_id == target_class_id:
|
||||
return {"message": f"该学生已在 '{target_class_name}' 班级中，无需重复操作"}
|
||||
|
||||
# 4. 逻辑删除原班级关联 (如果存在)
|
||||
if current_link:
|
||||
current_link.is_deleted = 1 # 标记为已删除
|
||||
current_link.update_time = now # 更新时间
|
||||
# 注意：这里不需要 db.add(current_link)，因为它是从 session查出来的对象，修改属性后 commit 即可生效
|
||||
|
||||
# 5. 新增新班级关联记录
|
||||
new_link = StudentClassIntermediate(
|
||||
id=str(uuid.uuid4())[:8], # 生成新的主键ID
|
||||
class_id=target_class_id, # 使用刚才查到的目标班级ID
|
||||
student_id=student_id,
|
||||
is_deleted=0, # 标记为有效
|
||||
create_time=now,
|
||||
update_time=now
|
||||
)
|
||||
db.add(new_link)
|
||||
|
||||
# 6. 提交事务
|
||||
db.commit()
|
||||
|
||||
return {
|
||||
"message": f"学生 {student_id} 成功转入班级 '{target_class.class_name}'",
|
||||
"new_class_id": target_class_id,
|
||||
"old_class_id": current_link.class_id if current_link else None
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
from model.class_model import Class_info
|
||||
|
||||
def add_class_dao(o,db):
|
||||
try:
|
||||
o1 = Class_info( **o) # 将字典解析成键值对关键字参数输入 title = xxx, userid = XXX
|
||||
db.add(o1)
|
||||
except:
|
||||
db.rollback()
|
||||
return False
|
||||
else:
|
||||
db.commit()
|
||||
return True
|
||||
|
||||
def update_class_dao(class_id,update_data,db):
|
||||
try: # 数据更新是有返回值的，返回的是影响的行数
|
||||
rows = db.query( Class_info ).filter( Class_info.class_id == class_id ).update( update_data )
|
||||
except:
|
||||
db.rollback()
|
||||
return False
|
||||
else:
|
||||
db.commit()
|
||||
return rows
|
||||
|
||||
|
||||
def get_class_dao(class_name,db):
|
||||
try:
|
||||
rows = db.query(Class_info).filter(Class_info.class_name == class_name).filter(Class_info.is_deleted==0).all()
|
||||
except:
|
||||
db.rollback()
|
||||
return False
|
||||
else:
|
||||
db.commit()
|
||||
return rows
|
||||
|
||||
|
||||
|
||||
class Student_delect_Dao:
|
||||
@staticmethod
|
||||
def soft_delete_student(db, class_id: str):
|
||||
# 先查询is_deleted为0的有效学生数据
|
||||
query = db.query(Class_info).where(
|
||||
Class_info.class_id == class_id,
|
||||
Class_info.is_deleted == 0
|
||||
)
|
||||
result = db.execute(query)
|
||||
target_student = result.scalar_one_or_none()
|
||||
return target_student
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
+18
@@ -0,0 +1,18 @@
|
||||
from sqlalchemy import *
|
||||
from sqlalchemy.orm import declarative_base,sessionmaker
|
||||
|
||||
db_url = "mysql+pymysql://root:123456@192.168.5.8:3306/student_system?charset=utf8mb4"
|
||||
engine = create_engine(db_url)
|
||||
|
||||
Base = declarative_base()
|
||||
Session = sessionmaker(bind=engine
|
||||
,autoflush=False
|
||||
,autocommit = False
|
||||
)
|
||||
|
||||
def get_db():
|
||||
db = Session()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
@@ -0,0 +1,21 @@
|
||||
from fastapi import FastAPI
|
||||
from database import engine, Base
|
||||
Base.metadata.create_all(engine)
|
||||
from api.calss import class_api
|
||||
from starlette.middleware.cors import CORSMiddleware
|
||||
app = FastAPI()
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"], # 允许所有源（生产环境建议指定具体域名）
|
||||
allow_credentials=True, # 允许携带 Cookie 等凭证
|
||||
allow_methods=["*"], # 允许所有 HTTP 方法（GET/POST/PUT/DELETE 等）
|
||||
allow_headers=["*"], # 允许所有请求头
|
||||
)
|
||||
|
||||
app.include_router(class_api,tags=['班级接口'],prefix='/stutent')
|
||||
|
||||
if __name__ == '__main__':
|
||||
import uvicorn
|
||||
|
||||
uvicorn.run("main:app", host='127.0.0.1', port=12345)
|
||||
@@ -0,0 +1,122 @@
|
||||
from database import Base
|
||||
from datetime import datetime,date
|
||||
from sqlalchemy import *
|
||||
from sqlalchemy import Column, Integer, String, Date, JSON
|
||||
from sqlalchemy import Column, String, Date, Enum, DateTime, func
|
||||
from sqlalchemy.orm import declarative_base
|
||||
import enum
|
||||
|
||||
class Class_info(Base): # python里的表名字
|
||||
__tablename__ = 'class_info' # 实际数据库的表名字
|
||||
class_id = Column(String(20)
|
||||
, primary_key=True
|
||||
, autoincrement=False
|
||||
, comment='班级编号，字符串类型手动维护')
|
||||
|
||||
class_code = Column(String(20) # 对应数据库的varchar类型
|
||||
, nullable=False
|
||||
,comment='班级业务编码，格式示例：2026-CS-01'# 不允许为null
|
||||
)
|
||||
class_name = Column(String(50)
|
||||
, nullable=False
|
||||
,comment='所在班级显示名称，如“2026级计算机科学与技术1班”，直接用于前端展示'
|
||||
)
|
||||
grade_year = Column(Date
|
||||
,default=date.today
|
||||
,comment='入学年级，如2026，用数值存储比字符串更便于按年级筛选、排序'
|
||||
)
|
||||
|
||||
# teacher_id = Column(Integer
|
||||
# , nullable=False
|
||||
# ,comment='班主任外键，关联教师表，一个班级对应一名班主任'
|
||||
# )
|
||||
#
|
||||
# student_id = Column(Integer
|
||||
# , nullable=False
|
||||
# ,comment='学号 用来连接学生表'
|
||||
# )
|
||||
|
||||
enrollment_date = Column(Date
|
||||
,default=date.today
|
||||
,comment='班级正式入学日期'
|
||||
)
|
||||
|
||||
status = Column(Integer
|
||||
, nullable=False
|
||||
,comment='班级正式入学日期'
|
||||
)
|
||||
|
||||
|
||||
tags = Column(JSON
|
||||
, nullable=False
|
||||
,comment='存储多维标签，例如：["重点班", "科技特长", "2026届"]'
|
||||
)
|
||||
|
||||
is_deleted = Column(Boolean, default=0, comment="0=数据存在 1=数据逻辑删除")
|
||||
|
||||
|
||||
create_time = Column(DATETIME
|
||||
, default=datetime.now # 默认值是当前时间
|
||||
)
|
||||
update_time = Column(DATETIME
|
||||
, default=datetime.now # 首次创建的时间跟更新时间一致
|
||||
, onupdate=datetime.now
|
||||
)
|
||||
|
||||
class StudentClassIntermediate(Base):
|
||||
__tablename__ = "student_class_intermediate_table"
|
||||
id = Column(String(20), primary_key=True)
|
||||
class_id= Column(String(20))
|
||||
student_id =Column(String(20))
|
||||
is_deleted= Column(Integer)
|
||||
create_time= Column(DateTime)
|
||||
update_time= Column(DateTime)
|
||||
|
||||
|
||||
|
||||
|
||||
class GenderEnum(str, enum.Enum):
|
||||
FEMALE = '0'
|
||||
MALE = '1'
|
||||
|
||||
class StudentStatusEnum(str, enum.Enum):
|
||||
NORMAL = '0' # 正常
|
||||
SUSPENDED = '1' # 休学
|
||||
DROPPED = '2' # 退学
|
||||
GRADUATED = '3' # 毕业
|
||||
OTHER = '4' # 其他
|
||||
|
||||
class EducationLevelEnum(str, enum.Enum):
|
||||
COLLEGE = '1' # 大专
|
||||
UNDERGRADUATE = '2' # 本科
|
||||
MASTER = '3' # 硕士
|
||||
DOCTOR = '4' # 博士
|
||||
|
||||
class IsDeletedEnum(str, enum.Enum):
|
||||
ACTIVE = '0' # 未删除
|
||||
DELETED = '1' # 已删除
|
||||
|
||||
# 3. 定义学生模型类
|
||||
class Student_info(Base):
|
||||
__tablename__ = "student_info"
|
||||
student_id = Column(String(20),primary_key=True, comment='学号')
|
||||
student_name = Column(String(50), nullable=False, comment='学生姓名')
|
||||
gender = Column(Enum(GenderEnum), nullable=False, comment='性别: 0=女, 1=男')
|
||||
id_card = Column(String(18), nullable=False, comment='身份证号码，收紧至18位')
|
||||
birthday = Column(Date, nullable=True, comment='出生日期')
|
||||
ethnicity = Column(String(20), nullable=True, comment='民族')
|
||||
region_id = Column(String(12), nullable=True, comment='籍贯地区编码，外键关联地区维度表(region_dimension)')
|
||||
phone = Column(String(11), nullable=True, comment='手机号码，收紧至11位')
|
||||
graduation_school = Column(String(20), nullable=True, comment='毕业学校')
|
||||
major = Column(String(20), nullable=False, comment='就读专业')
|
||||
class_id = Column(String(20), nullable=False, index=True, comment='所在班级，外键关联班级表(class_info)')
|
||||
enrollment_date = Column(Date, nullable=True, comment='入学日期')
|
||||
graduation_date = Column(Date, nullable=True, comment='毕业日期')
|
||||
student_status = Column(Enum(StudentStatusEnum), nullable=False, default=StudentStatusEnum.NORMAL,
|
||||
comment='学籍状态')
|
||||
education_level = Column(Enum(EducationLevelEnum), nullable=False,
|
||||
comment='学历层次：1=大专, 2=本科, 3=硕士研究生, 4=博士研究生')
|
||||
is_deleted = Column(Enum(IsDeletedEnum), nullable=False, default=IsDeletedEnum.ACTIVE,
|
||||
comment='逻辑删除：0=未删除, 1=已删除')
|
||||
create_time = Column(DateTime, nullable=False, server_default=func.now(), comment='创建时间')
|
||||
update_time = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now(), comment='更新时间')
|
||||
@@ -0,0 +1,36 @@
|
||||
from pydantic import BaseModel
|
||||
from typing import List
|
||||
|
||||
class ClassRequest(BaseModel):
|
||||
class_code: str | None = None
|
||||
class_id: str | None = None
|
||||
class_name: str | None = None
|
||||
tags: list[str] | None
|
||||
status: int| None
|
||||
|
||||
class ClassResponse(BaseModel):
|
||||
class_id: str
|
||||
class_name: str
|
||||
class_code: str
|
||||
student_count: int = 0
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
#同步转班
|
||||
class TransferRequest(BaseModel):
|
||||
target_class_name: str
|
||||
|
||||
|
||||
#分页模型
|
||||
class StudentDetailResponse(BaseModel):
|
||||
student_id: str
|
||||
student_name: str
|
||||
|
||||
|
||||
class ClassResponse(BaseModel):
|
||||
code: int = 200
|
||||
detail: str = 'OK'
|
||||
totals: int = 0
|
||||
data: str | dict | tuple | list
|
||||
|
||||
Reference in New Issue
Block a user