100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from model.wl_emp_model import Emp
|
|
from model.wl_student_model import Student
|
|
from scheme.wl_emp_scheme import EmpCreate, EmpUpdate
|
|
|
|
|
|
class EmpDAO:
|
|
|
|
@staticmethod
|
|
def get_student(db: Session, stu_no: str):
|
|
"""按学号查未删除的学生"""
|
|
return db.query(Student).filter(
|
|
Student.stu_no == stu_no,
|
|
Student.is_deleted == 0,
|
|
).first()
|
|
|
|
@staticmethod
|
|
def list_emps(db: Session, skip: int = 0, limit: int = 100, company_name: str = None):
|
|
"""查询就业列表，顺带把学生姓名带上"""
|
|
q = db.query(Emp).filter(Emp.is_deleted == 0)
|
|
if company_name:
|
|
q = q.filter(Emp.company_name.like(f"%{company_name}%"))
|
|
|
|
emp_list = q.offset(skip).limit(limit).all()
|
|
for e in emp_list:
|
|
stu = db.query(Student).filter(
|
|
Student.stu_no == e.stu_no,
|
|
Student.is_deleted == 0,
|
|
).first()
|
|
e.stu_name = stu.stu_name if stu else None
|
|
return emp_list
|
|
|
|
@staticmethod
|
|
def get_emp(db: Session, stu_no: str):
|
|
"""按学号查就业信息"""
|
|
emp = db.query(Emp).filter(
|
|
Emp.stu_no == stu_no,
|
|
Emp.is_deleted == 0,
|
|
).first()
|
|
if emp:
|
|
stu = EmpDAO.get_student(db, stu_no)
|
|
emp.stu_name = stu.stu_name if stu else None
|
|
return emp
|
|
|
|
@staticmethod
|
|
def create_emp(db: Session, emp_data: EmpCreate):
|
|
"""新增就业信息；如果这个学生之前被删过，就把那条记录复活"""
|
|
# stu_no 是主键，已删除的记录也占着这个主键，所以查的时候要连已删除的一起查
|
|
exist = db.query(Emp).filter(Emp.stu_no == emp_data.stu_no).first()
|
|
if exist and exist.is_deleted == 0:
|
|
raise ValueError("该学生的就业信息已存在")
|
|
|
|
# exclude_none：没传的字段不写进去，交给数据库默认值
|
|
data = emp_data.model_dump(exclude_none=True)
|
|
stu = EmpDAO.get_student(db, emp_data.stu_no)
|
|
|
|
if exist:
|
|
# 复活旧记录
|
|
for k, v in data.items():
|
|
setattr(exist, k, v)
|
|
exist.is_deleted = 0
|
|
db.commit()
|
|
db.refresh(exist)
|
|
exist.stu_name = stu.stu_name if stu else None
|
|
return exist
|
|
|
|
db_emp = Emp(**data)
|
|
db.add(db_emp)
|
|
db.commit()
|
|
db.refresh(db_emp)
|
|
db_emp.stu_name = stu.stu_name if stu else None
|
|
return db_emp
|
|
|
|
@staticmethod
|
|
def update_emp(db: Session, stu_no: str, emp_data: EmpUpdate):
|
|
"""修改就业信息"""
|
|
db_emp = EmpDAO.get_emp(db, stu_no)
|
|
if not db_emp:
|
|
return None
|
|
|
|
update_dict = emp_data.model_dump(exclude_unset=True)
|
|
for k, v in update_dict.items():
|
|
setattr(db_emp, k, v)
|
|
|
|
db.commit()
|
|
db.refresh(db_emp)
|
|
return db_emp
|
|
|
|
@staticmethod
|
|
def delete_emp(db: Session, stu_no: str):
|
|
"""逻辑删除，修改 is_deleted=1"""
|
|
db_emp = EmpDAO.get_emp(db, stu_no)
|
|
if not db_emp:
|
|
return None
|
|
|
|
db_emp.is_deleted = 1
|
|
db.commit()
|
|
return db_emp
|