学生信息管理系统

sqlalchemy_fastapi_demo/dao/advisor/crud.py

@@ -0,0 +1,56 @@
+from sqlalchemy.orm import Session
+from model.advisors import Advisor
+from model.students import Student
+
+def create_advisor(db:Session,advisor_id:int,advisor_name:str):
+    advisor = Advisor(advisor_id=advisor_id,advisor_name=advisor_name)
+    db.add(advisor)
+    db.commit()
+    db.refresh(advisor)
+    return advisor
+
+
+def search_advisor(db:Session,advisor_id:int|None=None,advisor_name:str|None=None):
+    if advisor_id is None and advisor_name is None:
+        return None
+    query=db.query(Advisor)
+    if advisor_id is not None:
+        query = query.filter(Advisor.advisor_id == advisor_id)
+    if advisor_name is not None:
+        query = query.filter(Advisor.advisor_name==advisor_name)
+    advisor = query.first()
+    return advisor
+
+
+def list_advisors(db: Session,advisor_name: str|None =None):
+    query = db.query(Advisor)
+    if advisor_name is not None:
+        query = query.filter(Advisor.advisor_name.like(f"%{advisor_name}%"))
+    return query.all()
+
+
+def list_students_of_advisor(db: Session, advisor_id: int):
+    """查这个顾问名下还没被逻辑删除的学生"""
+    return db.query(Student).filter(
+        Student.is_deleted == 0,
+        Student.advisor_id == advisor_id,
+    ).all()
+
+
+def update_advisor(db:Session,advisor_id:int,advisor_name:str|None=None):
+    r_advisor = search_advisor(db=db,advisor_id=advisor_id)
+    if r_advisor is None:
+        return None
+    if advisor_name is not None:
+        r_advisor.advisor_name = advisor_name
+    db.commit()
+    return r_advisor
+
+
+def delete_advisor(db:Session,advisor_id:int):
+    d_advisor = search_advisor(db=db,advisor_id=advisor_id)
+    if d_advisor is None:
+        return None
+    db.delete(d_advisor)
+    db.commit()
+    return advisor_id

sqlalchemy_fastapi_demo/dao/advisor/service.py

@@ -0,0 +1,41 @@
+from sqlalchemy.orm import Session
+from sqlalchemy.exc import IntegrityError
+from dao.advisor import crud
+from scheme.advisors import AdvisorIn
+
+# 成功 (数据, None)；失败 (None, "错误码")。
+
+def create_advisor(db:Session,advisor_id:int,advisor_name:str):
+    exist=crud.search_advisor(db,advisor_id=advisor_id)
+    if exist is not None:
+        return None,"DUPLICATE"  #ID主键已被占用
+    try:
+        advisor = crud.create_advisor(db,advisor_id,advisor_name)
+    except IntegrityError:
+        db.rollback()
+        return None, "DUPLICATE"
+    return advisor, None
+
+
+def update_advisor(db:Session,advisor_id:int,advisor_in:AdvisorIn):
+    exist = crud.search_advisor(db,advisor_id=advisor_id)
+    if exist is None:
+        return None, "NOT_FOUND"
+    u_advisor=crud.update_advisor(db,advisor_id=advisor_id,advisor_name=advisor_in.advisor_name)
+    if u_advisor is None:
+        return None, "NOT_FOUND"      # 查过之后被并发删掉了
+    return u_advisor,None
+
+
+#删除成功返回Advisor ID，删除失败返回None。顾问如连接学生，则拒绝删除。
+def delete_advisor(db: Session, advisor_id: int):
+    exist = crud.search_advisor(db=db, advisor_id=advisor_id)
+    if exist is None:
+        return None, "NOT_FOUND"
+    # 查一下这个顾问名下有没有学生
+    if crud.list_students_of_advisor(db, advisor_id):
+        return None, "HAS_STUDENTS"
+    result = crud.delete_advisor(db=db, advisor_id=advisor_id)
+    if result is None:
+        return None, "NOT_FOUND"      # 查过之后被并发删掉了
+    return result, None