第一版_完整
This commit is contained in:
jianqi
@@ -0,0 +1,218 @@
|
||||
# dao/employ_dao.py
|
||||
# 本文件封装 stu_emp_mgmt 表的所有数据库操作（增、删、改、查）
|
||||
# author:王博
|
||||
from datetime import date
|
||||
from sqlalchemy import and_
|
||||
from sqlalchemy.orm import Session
|
||||
from model.employ_model import StudentEmployManage as ETable
|
||||
from scheme.employ_scheme import EmployStatusCreate as EModel, EmployStatusQuery as QModel,EmployStatusDelete as DModel
|
||||
from model.stu_model import StuInfo
|
||||
|
||||
from typing import Optional, List
|
||||
class StudentEmployManageDAO:
|
||||
"""
|
||||
封装stu_emp_mgmt 表的所有数据库操作（增、删、改、查）
|
||||
"""
|
||||
@staticmethod
|
||||
def add_emp_reco(db:Session,emp_sta_add:EModel):
|
||||
"""
|
||||
登记学生就业信息
|
||||
:param db: 数据库会话
|
||||
:param emp_sta_add: 参数，对象，类型属于 学生就业管理表 模型类
|
||||
:return: 添加的stu_emp_info对象
|
||||
"""
|
||||
d_stu = emp_sta_add.model_dump() # 将请求体参数对象 ，转为字典
|
||||
if d_stu['emp_open_time']:
|
||||
d_stu['emp_open_time'] = date(**d_stu['emp_open_time']) # 就业开放时间
|
||||
if d_stu['send_offer_time']:
|
||||
d_stu['send_offer_time'] = date(**d_stu['send_offer_time']) # offer下发时间
|
||||
db_emp = ETable(**d_stu)
|
||||
db.add(db_emp)
|
||||
db.commit()
|
||||
db.refresh(db_emp)
|
||||
return db_emp
|
||||
|
||||
|
||||
@staticmethod
|
||||
def is_in_stu(db: Session, emp_sta_in: EModel):
|
||||
"""
|
||||
判断 参数中 学生id 是否在 学生表StuInfo 中存在
|
||||
根据 学生id、is_deleted 过滤
|
||||
:return: 存在 返回True；不存在返回False
|
||||
"""
|
||||
stu = db.query(StuInfo).filter(and_(StuInfo.id==emp_sta_in.stu_id,
|
||||
StuInfo.is_deleted==0)).first()
|
||||
if stu: # 能找到学号，并且 是否 删除状态为 0
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def is_exist_visible(db: Session, emp_sta_exist:List[DModel] |EModel | QModel | str):
|
||||
"""
|
||||
判断学生就业信息是否在 学生就业信息表中 存在
|
||||
is_deleted 0，用来判断可见、可查询
|
||||
:return: 存在、可见 返回True；否则 返回False
|
||||
"""
|
||||
emp = None
|
||||
emp_info = db.query(ETable)
|
||||
if isinstance(emp_sta_exist,str):
|
||||
emp = emp_info.filter(and_(ETable.stu_id == emp_sta_exist,
|
||||
ETable.is_deleted == 0)).first()
|
||||
|
||||
elif isinstance(emp_sta_exist, list):
|
||||
for i in emp_sta_exist:
|
||||
emp = emp_info.filter(and_(ETable.stu_id == i.stu_id,
|
||||
ETable.is_deleted == 0)).first()
|
||||
if not emp:
|
||||
emp = False
|
||||
break
|
||||
|
||||
else:
|
||||
emp = emp_info.filter(and_(ETable.stu_id == emp_sta_exist.stu_id,
|
||||
ETable.is_deleted == 0)).first()
|
||||
if emp:
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def is_exist_invisible(db: Session, emp_sta_exist: EModel):
|
||||
"""
|
||||
判断学生就业信息是否在 学生就业信息表中 存在
|
||||
is_deleted 1，用来判断不可见、不可查询
|
||||
:return: 存在、不可见 返回True；否则 返回False
|
||||
"""
|
||||
emp_info = db.query(ETable).filter(and_(ETable.stu_id == emp_sta_exist.stu_id,
|
||||
ETable.is_deleted == 1)).first()
|
||||
if emp_info:
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def set_stu_state(db: Session, emp_sta_set: EModel):
|
||||
"""
|
||||
根据就业开放时间、offer下发时间 设置 学生就业状态
|
||||
就业未开放 ---》在读
|
||||
就业开放&offer未下发 ---》进入就业
|
||||
就业开放&offer下发 ---》已就业
|
||||
:return: 在读、进入就业、已就业
|
||||
"""
|
||||
stu = db.query(StuInfo).filter(StuInfo.id == emp_sta_set.stu_id).first()
|
||||
if emp_sta_set.emp_open_time is None: # 就业开放时间为 None
|
||||
stu.state = '在读'
|
||||
db.commit()
|
||||
db.refresh(stu)
|
||||
return stu.state
|
||||
|
||||
elif emp_sta_set.send_offer_time is None:# 有就业开放时间，offer下发时间为 None
|
||||
stu.state = '进入就业'
|
||||
db.commit()
|
||||
db.refresh(stu)
|
||||
return stu.state
|
||||
|
||||
else: # 有offer下发时间
|
||||
stu.state = '已就业'
|
||||
db.commit()
|
||||
db.refresh(stu)
|
||||
return stu.state
|
||||
|
||||
@staticmethod
|
||||
def set_emp_info(db: Session, emp_sta_update: EModel):
|
||||
"""
|
||||
根据请求体参数 修改 学生 就业信息
|
||||
:return: 修改条数0或1；0代表 修改失败 ；1代表 修改成功
|
||||
"""
|
||||
d_stu = emp_sta_update.model_dump() # 将请求体参数对象 ，转为字典
|
||||
if d_stu['emp_open_time']:
|
||||
d_stu['emp_open_time'] = date(**d_stu['emp_open_time']) # 转化成 就业开放时间
|
||||
if d_stu['send_offer_time']:
|
||||
d_stu['send_offer_time'] = date(**d_stu['send_offer_time']) # 转化成 offer下发时间
|
||||
row_count = db.query(ETable).filter(ETable.stu_id == emp_sta_update.stu_id).update(d_stu) #修改条数0或1
|
||||
db.commit()
|
||||
return row_count
|
||||
|
||||
@staticmethod
|
||||
def batch_set_emp_info(db: Session, emp_sta_update:EModel):
|
||||
"""
|
||||
根据请求体参数列表 批量修改 学生 就业信息
|
||||
:return: 修改条数0或1；0代表 修改失败 ；1代表 修改成功
|
||||
"""
|
||||
d_stu = emp_sta_update.model_dump() # 将请求体参数对象 ，转为字典
|
||||
if d_stu['emp_open_time']:
|
||||
d_stu['emp_open_time'] = date(**d_stu['emp_open_time']) # 转化成 就业开放时间
|
||||
if d_stu['send_offer_time']:
|
||||
d_stu['send_offer_time'] = date(**d_stu['send_offer_time']) # 转化成 offer下发时间
|
||||
row_count = db.query(ETable).filter(ETable.stu_id == emp_sta_update.stu_id).update(d_stu) # 修改条数0或1
|
||||
return row_count
|
||||
|
||||
|
||||
@staticmethod
|
||||
def is_deleted(db: Session, emp_info):
|
||||
if emp_info:
|
||||
emp_info.is_deleted = 1
|
||||
db.commit()
|
||||
db.refresh(emp_info)
|
||||
|
||||
@staticmethod
|
||||
def del_emp_info(db: Session, emp_sta_del: List[DModel] | str):
|
||||
"""
|
||||
根据请求体参数 逻辑删除 学生就业信息
|
||||
:return: 返回逻辑删除的 学生就业信息
|
||||
"""
|
||||
emp_info = db.query(ETable)
|
||||
|
||||
if isinstance(emp_sta_del, str):
|
||||
emp = emp_info.filter(and_(ETable.stu_id == emp_sta_del,
|
||||
ETable.is_deleted == 0)).first()
|
||||
StudentEmployManageDAO.is_deleted(db,emp)
|
||||
db.commit()
|
||||
return True
|
||||
if isinstance(emp_sta_del, list):
|
||||
emp_copy = emp_sta_del.copy()
|
||||
for i in emp_copy:
|
||||
emp = emp_info.filter(and_(ETable.stu_id == i.stu_id,
|
||||
ETable.is_deleted ==0)).first()
|
||||
StudentEmployManageDAO.is_deleted(db, emp)
|
||||
db.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def recover_emp_info(db: Session, emp_sta_recover: EModel):
|
||||
"""
|
||||
根据请求体参数 逻辑删除 学生就业信息
|
||||
:return: 返回逻辑删除的 学生就业信息
|
||||
"""
|
||||
emp_info = db.query(ETable).filter(and_(ETable.stu_id == emp_sta_recover.stu_id,
|
||||
ETable.is_deleted ==0)).first()
|
||||
if emp_info:
|
||||
emp_info.is_deleted = 0
|
||||
db.commit()
|
||||
db.refresh(emp_info)
|
||||
return emp_info
|
||||
|
||||
@staticmethod
|
||||
def query_emp_info(db: Session, emp_sta_query: QModel):
|
||||
"""
|
||||
请求体传入学生编号(可选)、就业公司(可选)、薪资范围(可选) 可多条件查询 学生就业信息
|
||||
:return: 返回查询到的 所有 学生就业信息
|
||||
"""
|
||||
e_all = db.query(ETable)
|
||||
if emp_sta_query.stu_id: # 根据 学生编号 筛选出 学生就业信息
|
||||
e_all = e_all.filter(ETable.stu_id == emp_sta_query.stu_id)
|
||||
if emp_sta_query.emp_company: # 根据 就业公司名称 模糊筛选出 学生就业信息
|
||||
e_all = e_all.filter(ETable.emp_company.like(f"%{emp_sta_query.emp_company}%"))
|
||||
if emp_sta_query.min_salary: # 根据 最小薪资范围 筛选出 大于 最小薪资范围的 学生就业信息
|
||||
e_all = e_all.filter(ETable.salary >= emp_sta_query.min_salary)
|
||||
if emp_sta_query.max_salary: # 根据 最大薪资范围 筛选出 小于 最大薪资范围的 学生就业信息
|
||||
e_all = e_all.filter(ETable.salary <= emp_sta_query.max_salary)
|
||||
|
||||
total = e_all.count()
|
||||
e_all = e_all.offset((emp_sta_query.skip-1)*emp_sta_query.limit).limit(emp_sta_query.limit).all()
|
||||
|
||||
return e_all,total
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
from sqlalchemy.orm import Session
|
||||
from model.stu_model import StuInfo
|
||||
from model.cls_mgmt_model import ClsMgmt
|
||||
from scheme.stu_scheme import StudentUpdate
|
||||
|
||||
|
||||
class StudentDao:
|
||||
"""查学生"""
|
||||
#---------分页查询所有学生信息----------
|
||||
@staticmethod
|
||||
def get_all(db: Session, skip: int = 0, limit: int = 20):
|
||||
return (db.query(StuInfo)
|
||||
.filter(StuInfo.is_deleted == 0)
|
||||
.offset(skip).limit(limit).all())
|
||||
#---------根据学生id查询学生信息----------
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, stu_id: str) :
|
||||
return (db.query(StuInfo).
|
||||
filter(StuInfo.id == stu_id)
|
||||
.filter(StuInfo.is_deleted == 0)
|
||||
.first())
|
||||
#---------根据学生姓名查询学生信息----------
|
||||
@staticmethod
|
||||
def get_by_name(db: Session, stu_name: str) :
|
||||
return (db.query(StuInfo)
|
||||
.filter(StuInfo.name == stu_name)
|
||||
.filter(StuInfo.is_deleted == 0)
|
||||
.all())
|
||||
#---------根据班级id查询班级学生信息----------
|
||||
@staticmethod
|
||||
def get_by_class(db: Session, cls_id: str) :
|
||||
return (db.query(StuInfo)
|
||||
.filter(StuInfo.cls_id == cls_id)
|
||||
.filter(StuInfo.is_deleted == 0)
|
||||
.all())
|
||||
# """增加学生"""
|
||||
# 1 ---------添加学生信息----------
|
||||
@staticmethod
|
||||
def create(db: Session, student_data: dict):
|
||||
db_student=StuInfo(**student_data)
|
||||
db.add(db_student)
|
||||
db.commit()
|
||||
db.refresh(db_student) # 刷新对象，获取数据库生成的默认值（如 created_at）
|
||||
return db_student
|
||||
# 2 ---------查询班级是否存在(可能班级创建了但还没人)----------
|
||||
@staticmethod
|
||||
def get_class_by_id(db: Session,cls_id:str):
|
||||
return (db.query(ClsMgmt)
|
||||
.filter(ClsMgmt.id==cls_id)
|
||||
.first()) #不存在自动返回None
|
||||
# 3 ---------查询对应的班级人数用来生成学号----------
|
||||
@staticmethod
|
||||
def count_by_class(db: Session, cls_id: str) -> int:
|
||||
return (db.query(StuInfo)
|
||||
.filter(StuInfo.cls_id == cls_id)
|
||||
.filter(StuInfo.is_deleted == 0)
|
||||
.count())
|
||||
# """修改学生信息"""
|
||||
#---------根据学生id查找并修改学生信息----------
|
||||
@staticmethod
|
||||
def update(db:Session,stu_id:str,stu_data:StudentUpdate):
|
||||
db_student = StudentDao.get_by_id(db,stu_id)
|
||||
if not db_student:
|
||||
return None
|
||||
# model_dump把用户传入修改请求体的信息转成字典，exclude_unset=True把没传入的字段排除掉，否则会None来修改其他字段
|
||||
update_data = stu_data.model_dump(exclude_unset=True)
|
||||
for key, value in update_data.items(): # 动态设置属性
|
||||
setattr(db_student, key, value)
|
||||
|
||||
db.commit() # 提交事务
|
||||
db.refresh(db_student) # 刷新对象，获取 onupdate 时间等
|
||||
return db_student
|
||||
# """删除学生信息""" 逻辑删除
|
||||
#---------根据学生id查找并删除学生信息----------
|
||||
@staticmethod
|
||||
def delete(db:Session,stu_id:str):
|
||||
db_student=StudentDao.get_by_id(db,stu_id)
|
||||
if not db_student:
|
||||
return False
|
||||
db_student.is_deleted=1
|
||||
db.commit()
|
||||
db.refresh(db_student)
|
||||
return db_student
|
||||
@@ -0,0 +1,109 @@
|
||||
# dao/stu_score_dao.py
|
||||
# 本文件封装对 User 表的所有数据库操作（增、删、改、查）
|
||||
from sqlalchemy import and_
|
||||
from sqlalchemy.orm import Session
|
||||
from model.stu_score_model import StuScore
|
||||
# from scheme.stu_score_scheme import
|
||||
from typing import Optional, List
|
||||
|
||||
class StuScoreDAO:
|
||||
"""用户数据访问对象，所有方法均为静态方法，方便调用"""
|
||||
|
||||
# # ------------------增添手段----------------------
|
||||
@staticmethod
|
||||
def add_score(db: Session,stu_id: str,exam_attempt:int,exam_score:int,score_level:str,is_deleted: bool
|
||||
) -> StuScore:
|
||||
new_score = StuScore(stu_id=stu_id,
|
||||
exam_attempt=exam_attempt,
|
||||
exam_score=exam_score,
|
||||
score_level=score_level,
|
||||
is_deleted=is_deleted)
|
||||
db.add(new_score)
|
||||
db.commit()
|
||||
return new_score
|
||||
|
||||
# ------------------查询手段----------------------
|
||||
@staticmethod
|
||||
def get_all_score(db: Session, skip: int = 0, limit: int = 10) -> List[StuScore]:
|
||||
"""
|
||||
获取所有用户（支持分页）
|
||||
:param db: 数据库会话
|
||||
:param skip: 偏移量（跳过前 skip 条）
|
||||
:param limit: 最大返回条数
|
||||
:return: 用户对象列表
|
||||
"""
|
||||
return db.query(StuScore).filter(StuScore.is_deleted == False).offset(skip).limit(limit).all()
|
||||
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, stu_id: str) -> List[StuScore]:
|
||||
"""
|
||||
根据学生ID 获取成绩
|
||||
:return: 用户对象或 None
|
||||
"""
|
||||
return db.query(*StuScoreDAO.get_field()).filter(and_(StuScore.stu_id == stu_id,StuScore.is_deleted == False)).all()
|
||||
|
||||
@staticmethod
|
||||
def get_by_exam_attempt(db: Session, exam_attempt: int) -> List[StuScore]:
|
||||
"""
|
||||
根据考试轮次 获取成绩
|
||||
:return: 用户对象或 None
|
||||
"""
|
||||
return db.query(*StuScoreDAO.get_field()).filter(and_(StuScore.exam_attempt == exam_attempt,StuScore.is_deleted == False)).all()
|
||||
|
||||
@staticmethod
|
||||
def get_field():
|
||||
get_field = StuScore.exam_score,StuScore.score_level,StuScore.stu_id,StuScore.exam_attempt
|
||||
return get_field
|
||||
|
||||
@staticmethod
|
||||
def get_by_id_attempt(db: Session, stu_id: str,exam_attempt: int) -> Optional[StuScore]:
|
||||
"""
|
||||
根据学生ID 和考试轮次精确定位成绩
|
||||
:return: 用户对象或 None
|
||||
"""
|
||||
return db.query(*StuScoreDAO.get_field()).filter(and_(StuScore.stu_id == stu_id , StuScore.exam_attempt == exam_attempt,StuScore.is_deleted == False)).first()
|
||||
|
||||
@staticmethod
|
||||
def get_by_id_attempt_all(db: Session, stu_id: str, exam_attempt: int) -> Optional[StuScore]:
|
||||
"""
|
||||
根据学生ID 和考试轮次精确定位成绩
|
||||
:return: 用户对象或 None
|
||||
"""
|
||||
return db.query(StuScore).filter(
|
||||
and_(StuScore.stu_id == stu_id, StuScore.exam_attempt == exam_attempt,
|
||||
StuScore.is_deleted == False)).first()
|
||||
|
||||
# ------------------更新手段----------------------
|
||||
@staticmethod
|
||||
def update_score(db: Session, stu_id: str,exam_attempt:int,exam_score:int,score_leval) -> Optional[StuScore]:
|
||||
|
||||
|
||||
db_score = StuScoreDAO.get_by_id_attempt_all(db,stu_id, exam_attempt)
|
||||
if not db_score:
|
||||
return None
|
||||
db_score.exam_score = exam_score
|
||||
db_score.score_level = score_leval
|
||||
|
||||
db.commit() # 提交事务
|
||||
db.refresh(db_score) # 刷新对象，获取 onupdate 时间等
|
||||
return db_score
|
||||
|
||||
# ------------------删除手段----------------------
|
||||
@staticmethod
|
||||
def delete_score(db: Session, stu_id: str,exam_attempt:int) -> bool:
|
||||
db_score = StuScoreDAO.get_by_id_attempt_all(db, stu_id, exam_attempt)
|
||||
if not db_score:
|
||||
return False
|
||||
db_score.is_deleted=True # 标记删除
|
||||
db.commit() # 提交事务
|
||||
return True
|
||||
@staticmethod
|
||||
def delete_all_score(db: Session, stu_id: str) -> bool:
|
||||
db_score = StuScoreDAO.get_by_id(db, stu_id)
|
||||
if not db_score:
|
||||
return False
|
||||
for score in db_score:
|
||||
score.is_delete=True # 标记删除
|
||||
db.commit() # 提交事务
|
||||
return True
|
||||
@@ -0,0 +1,146 @@
|
||||
# dao/teacher_dao.py
|
||||
# 本文件封装对 tes_info 表的所有数据库操作（增、删、改、查）
|
||||
from http.client import HTTPException
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlalchemy import or_
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from model.cls_mgmt_model import ClsMgmt
|
||||
from model.teacher_model import Teacher
|
||||
from scheme.teacher_scheme import TeacherUpdate, TeacherAdd
|
||||
import random
|
||||
|
||||
|
||||
class TeacherDao:
|
||||
"""教师数据访问对象，所有方法均为静态方法，方便调用"""
|
||||
|
||||
#获取所有教师基本信息
|
||||
@staticmethod
|
||||
def get_all(db: Session, skip: int = 0, limit: int = 100) :
|
||||
"""
|
||||
获取所有用户（支持分页）
|
||||
:param db: 数据库会话
|
||||
:param skip: 偏移量（跳过前 skip 条）
|
||||
:param limit: 最大返回条数
|
||||
:return: 教师对象列表
|
||||
"""
|
||||
return db.query(Teacher).filter(Teacher.is_deleted == 0).offset(skip).limit(limit).all()
|
||||
|
||||
|
||||
#获取单个教师信息
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, teacher_id: str) :
|
||||
"""
|
||||
根据主键 ID 获取单个教师信息
|
||||
:return:教师对象或 None
|
||||
"""
|
||||
return db.query(Teacher).filter(Teacher.id == teacher_id,Teacher.is_deleted == 0).first()
|
||||
|
||||
@staticmethod
|
||||
def get_by_name(db: Session, teacher_name: str) :
|
||||
"""
|
||||
根据教师姓名获取信息（用于唯一性检查）
|
||||
"""
|
||||
return db.query(Teacher).filter(Teacher.name == teacher_name,Teacher.is_deleted == 0).first()
|
||||
|
||||
# 根据教师id获取教师带教班级信息
|
||||
@staticmethod
|
||||
def get_clss_info(db: Session, teacher_id: str):
|
||||
"""
|
||||
根据主键 ID 获取单个教师信息
|
||||
:return: 带教信息或 None
|
||||
"""
|
||||
return db.query(ClsMgmt).filter(or_(ClsMgmt.head_tea_id == teacher_id,ClsMgmt.lecturer_id == teacher_id), ClsMgmt.is_deleted == 0).all()
|
||||
|
||||
|
||||
#添加教师
|
||||
@staticmethod
|
||||
def add(db: Session, teacher_data: TeacherAdd) :
|
||||
"""
|
||||
添加新教师
|
||||
:param db: 数据库会话
|
||||
:param teacher_data: 符合 TeacherAddUpdate 请求体模型的数据
|
||||
:return: 创建后的 teacher 对象（含id ）
|
||||
"""
|
||||
|
||||
# 教师id自动生成 格式:T+随机四位数 并进行重复验证
|
||||
while True:
|
||||
new_id = f"T{random.randint(100, 999)}" # 确保是4位数，不会出现 0012 这种情况
|
||||
|
||||
# 查询数据库是否已存在这个 id
|
||||
exists = db.query(Teacher).filter(Teacher.id == new_id).first()
|
||||
if not exists:
|
||||
break # 不存在，跳出循环，使用这个 id
|
||||
|
||||
# 把传入的参数变成字典
|
||||
data_dict = teacher_data.model_dump()
|
||||
|
||||
# 把生成id加到字典里
|
||||
data_dict["id"] = new_id
|
||||
|
||||
#软删除逻辑is_delete默认为0
|
||||
data_dict["is_deleted"] = 0
|
||||
|
||||
# 将 Pydantic 模型转为字典，并解包构建 SQLAlchemy 模型实例
|
||||
# 对字典进行拆包变成Teacher(id='',name='', phone='', type='',is_delete='0')
|
||||
db_teacher = Teacher(**data_dict)
|
||||
db.add(db_teacher) # 添加到会话
|
||||
db.commit() # 提交事务，此时会执行 INSERT，并自动填充自增字段
|
||||
db.refresh(db_teacher) # 刷新对象，获取数据库生成的默认值（如 created_at）
|
||||
return db_teacher
|
||||
|
||||
#更新教师信息
|
||||
@staticmethod
|
||||
def update(db: Session, teacher_id: str, teacher_data: TeacherUpdate) :
|
||||
"""
|
||||
更新教师信息（只更新传入的非空字段）
|
||||
:param db: 数据库会话
|
||||
:param teacher_id: 要更新的教师 ID
|
||||
:param teacher_data: 包含要更新字段的 Pydantic 模型
|
||||
:return: 更新后的 Teacher 对象，如果没有此教师则返回None
|
||||
"""
|
||||
db_teacher = TeacherDao.get_by_id(db, teacher_id)
|
||||
if not db_teacher:
|
||||
return None
|
||||
|
||||
# 只更新客户端显式传入的字段（exclude_unset=True 排除未设置的字段）
|
||||
# 不加 exclude_unset=True;teacher_data.model_dump() 会输出所有字段，没传的字段值是 None：
|
||||
# 将传入参数转变成字典
|
||||
update_data = teacher_data.model_dump(exclude_unset=True)
|
||||
#如果不传数据,直接返回本id教师
|
||||
if not update_data:
|
||||
return db_teacher
|
||||
|
||||
#拆分遍历传入的数据
|
||||
for key, value in update_data.items():
|
||||
#滚动插入数据 动态赋值
|
||||
setattr(db_teacher, key, value)
|
||||
|
||||
""""
|
||||
不使用setattr的写法
|
||||
if 'name' in update_data:
|
||||
db_teacher.name = update_data['name']
|
||||
if 'phone' in update_data:
|
||||
db_teacher.phone = update_data['phone']
|
||||
if 'type' in update_data:
|
||||
db_teacher.type = update_data['type']
|
||||
"""
|
||||
|
||||
db.commit() # 提交事务
|
||||
db.refresh(db_teacher) # 刷新对象，获取 onupdate 时间等
|
||||
return db_teacher
|
||||
|
||||
# 通过id删除教师
|
||||
@staticmethod
|
||||
def delete(db: Session, teacher_id: str):
|
||||
db_teacher = TeacherDao.get_by_id(db, teacher_id)
|
||||
#如果不存在返回None,供接口判断
|
||||
if not db_teacher:
|
||||
return None
|
||||
#存在的情况下,只需把is_delete改成1
|
||||
db_teacher.is_deleted = 1 #软删除,1代表已删除
|
||||
# db.delete(db_teacher) # 正常删除
|
||||
db.commit() # 提交事务
|
||||
db.refresh(db_teacher)
|
||||
return True
|
||||
@@ -1,86 +0,0 @@
|
||||
# dao/users_dao.py
|
||||
# 本文件封装对 User 表的所有数据库操作（增、删、改、查）
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
from model.users import User
|
||||
from scheme.users import UserCreate, UserUpdate
|
||||
from typing import Optional, List
|
||||
|
||||
class UserDAO:
|
||||
"""用户数据访问对象，所有方法均为静态方法，方便调用"""
|
||||
|
||||
@staticmethod
|
||||
def get_all(db: Session, skip: int = 0, limit: int = 100) -> List[User]:
|
||||
"""
|
||||
获取所有用户（支持分页）
|
||||
:param db: 数据库会话
|
||||
:param skip: 偏移量（跳过前 skip 条）
|
||||
:param limit: 最大返回条数
|
||||
:return: 用户对象列表
|
||||
"""
|
||||
return db.query(User).offset(skip).limit(limit).all()
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, user_id: int) -> Optional[User]:
|
||||
"""
|
||||
根据主键 ID 获取单个用户
|
||||
:return: 用户对象或 None
|
||||
"""
|
||||
return db.query(User).filter(User.id == user_id).first()
|
||||
|
||||
@staticmethod
|
||||
def get_by_username(db: Session, username: str) -> Optional[User]:
|
||||
"""
|
||||
根据用户名获取用户（用于唯一性检查）
|
||||
"""
|
||||
return db.query(User).filter(User.username == username).first()
|
||||
|
||||
@staticmethod
|
||||
def create(db: Session, user_data: UserCreate) -> User:
|
||||
"""
|
||||
创建新用户
|
||||
:param db: 数据库会话
|
||||
:param user_data: 符合 UserCreate 模型的数据
|
||||
:return: 创建后的 User 对象（含自增 id 和默认时间）
|
||||
"""
|
||||
# 将 Pydantic 模型转为字典，并解包构建 SQLAlchemy 模型实例
|
||||
db_user = User(**user_data.model_dump())
|
||||
db.add(db_user) # 添加到会话
|
||||
db.commit() # 提交事务，此时会执行 INSERT，并自动填充自增字段
|
||||
db.refresh(db_user) # 刷新对象，获取数据库生成的默认值（如 created_at）
|
||||
return db_user
|
||||
|
||||
@staticmethod
|
||||
def update(db: Session, user_id: int, user_data: UserUpdate) -> Optional[User]:
|
||||
"""
|
||||
更新用户信息（只更新传入的非空字段）
|
||||
:param db: 数据库会话
|
||||
:param user_id: 要更新的用户 ID
|
||||
:param user_data: 包含要更新字段的 Pydantic 模型
|
||||
:return: 更新后的 User 对象，如果用户不存在则返回 None
|
||||
"""
|
||||
db_user = UserDAO.get_by_id(db, user_id)
|
||||
if not db_user:
|
||||
return None
|
||||
|
||||
# 只更新客户端显式传入的字段（exclude_unset=True 排除未设置的字段）
|
||||
update_data = user_data.model_dump(exclude_unset=True)
|
||||
for key, value in update_data.items():
|
||||
setattr(db_user, key, value) # 动态设置属性
|
||||
|
||||
db.commit() # 提交事务
|
||||
db.refresh(db_user) # 刷新对象，获取 onupdate 时间等
|
||||
return db_user
|
||||
|
||||
@staticmethod
|
||||
def delete(db: Session, user_id: int) -> bool:
|
||||
"""
|
||||
删除用户
|
||||
:return: True 表示删除成功，False 表示用户不存在
|
||||
"""
|
||||
db_user = UserDAO.get_by_id(db, user_id)
|
||||
if not db_user:
|
||||
return False
|
||||
db.delete(db_user) # 标记删除
|
||||
db.commit() # 提交事务
|
||||
return True
|
||||
Reference in New Issue
Block a user