上传文件至「/」

2026-09-21 17:07:09 +08:00
parent 4476370353
commit 1743864918
5 changed files with 477 additions and 0 deletions
@@ -0,0 +1,132 @@
+
+from fastapi import Query, Depends, HTTPException, APIRouter
+from sqlalchemy.orm import Session
+from sqlalchemy_fastapi_demo.dao.Score import ScoreDAO
+from sqlalchemy_fastapi_demo.database import get_db
+from sqlalchemy_fastapi_demo.scheme.Score import ScoreCreate, ScoreUpdate
+
+app_score=APIRouter()
+
+# 查询所有成绩
+@app_score.get("/",summary='查询所有成绩')
+async def get_scores(
+    skip: int = Query(0, ge=0, description="跳过的记录数"),
+    limit: int = Query(100, ge=1, le=200, description="返回的最大记录数"),
+    db: Session = Depends(get_db)
+):
+    scores=ScoreDAO.get_all(db, skip=skip, limit=limit)
+    return scores
+
+# 查询单个数据
+@app_score.get("/{stu_id}/{exam_id}",summary='查询单个成绩')
+async def get_score(
+        stu_id: int,
+        exam_id: int,
+        db: Session = Depends(get_db)
+):
+    score=ScoreDAO.get_by_id(db, stu_id,exam_id)
+    if not score:
+        raise HTTPException(status_code=404, detail="学生不存在")
+    return score
+
+# 创建学生
+@app_score.post("/",summary='创建学生成绩')
+async def create_scores(
+        score:ScoreCreate,
+        db: Session = Depends(get_db)
+):
+    # 检查学生在不在学生表
+    existing_foreign_key=ScoreDAO.inspect_stu_id_unq(db, score.stu_id)
+    if not existing_foreign_key:
+        raise HTTPException(status_code=404, detail="该学生不存在")
+    if existing_foreign_key.is_deleted == 1:
+        raise HTTPException(status_code=404, detail="该学生不存在")
+    # 检查成绩是否已被占用
+    existing_score=ScoreDAO.get_by_id(db, score.stu_id,score.exam_id)
+    if existing_score:
+        raise HTTPException(status_code=400, detail="学生成绩已存在，请勿重复添加")
+    new_score=ScoreDAO.post_score(db,score)
+    return new_score
+
+# 修改
+@app_score.put("/{stu_id}/{exam_id}",summary='修改信息')
+async def update_scores(
+        stu_id: int,
+        exam_id: int,
+        score:ScoreUpdate,
+        db: Session = Depends(get_db)
+):
+    # 只允许修改成绩，学生id和考试次序不允许修改，SQLAlchemy不能直接更新联合主键本身
+    # 检查成绩是否已存在
+    existing_foreign_key = ScoreDAO.inspect_stu_id_unq(db, score.stu_id)
+    if not existing_foreign_key:
+        raise HTTPException(status_code=404, detail="该学生不存在")
+    existing_score=ScoreDAO.get_by_id(db, stu_id,exam_id)
+    if not existing_score:
+        raise HTTPException(status_code=404, detail="学生成绩不存在")
+    # 直接调用DAO更新，只更新score字段
+    b = ScoreDAO.put_score(db, stu_id, exam_id, score)
+    return b
+
+# 修改成绩：支持修改 stu_id、exam_id、score；原理：删除旧记录，新增新记录
+    # stu_id、exam_id是联合主键。MySQL不允许update 直接修改主键字段，SQLAlchemy
+    # 底层执行会抛数据库异常。所以采用「删旧、建新」的方案模拟修改
+
+    # 风险大
+    # 现在代码：删旧和新增是两次
+    # db.commit()。
+    # 如果删旧成功，新增中途崩掉，会出现旧数据没了，新数据没加上，数据丢失。
+
+    # 数据库这条记录实际上是新行，不是原来那一行。
+    # 如果别的表有外键引用这条成绩记录，这个方案会出问题（旧记录删掉，关联跟着没了）。
+# @app_score.put("/{stu_id}/{exam_id}",summary='修改学生成绩(支持修改学生ID、考核序次、分数)')
+# async def update_scores(
+#         stu_id: int,
+#         exam_id: int,
+#         score:ScoreUpdate,
+#         db: Session = Depends(get_db)
+# ):
+#     # 1.找到旧记录
+#     old_score = ScoreDAO.get_by_id(db, stu_id, exam_id)
+#     if not old_score:
+#         raise HTTPException(status_code=404, detail="待修改的成绩记录不存在")
+#     # 2.确定新值：前端没传的字段，沿用旧记录的值
+#     new_stu = score.stu_id if score.stu_id is not None else old_score.stu_id
+#     new_exam_id = score.exam_id if score.exam_id is not None else old_score.exam_id
+#     new_score_val = score.score if score.score is not None else old_score.score
+#     # 3.校验新stu_id学生是否存在
+#     if new_stu != old_score.stu_id:
+#         stu_obj = ScoreDAO.inspect_stu_id_unq(db, new_stu)
+#         if not stu_obj:
+#             raise HTTPException(status_code=404, detail="目标学生不存在")
+#     # 4.如果新主键组合和旧的不一样，要检查新组合是否冲突
+#     if (new_stu, new_exam_id) != (old_score.stu_id, old_score.exam_id):
+#         conflict = ScoreDAO.get_by_id(db, new_stu, new_exam_id)
+#         if conflict:
+#             raise HTTPException(status_code=400, detail="新【学生ID+考核序次】组合已存在，不能使用")
+#     # 核心逻辑：删除旧记录，新增一条全新记录
+#     # 删除旧数据
+#     ScoreDAO.delete_score(db, stu_id, exam_id)
+#     # 组装成ScoreCreate对象，调用新增DAO
+#     from sqlalchemy_fastapi_demo.scheme.Score import ScoreCreate
+#     new_create_data = ScoreCreate(
+#         stu_id=new_stu,
+#         exam_id=new_exam_id,
+#         score=new_score_val
+#     )
+#     new_db_score = ScoreDAO.post_score(db, new_create_data)
+#     return new_db_score
+
+
+
+# 删除
+@app_score.delete("/{stu_id}/{exam_id}",summary='删除学生成绩')
+async def delete_scores(
+        stu_id: int,
+        exam_id: int,
+        db: Session = Depends(get_db)
+):
+    score=ScoreDAO.delete_score(db, stu_id,exam_id)
+    if not score:
+        raise HTTPException(status_code=404, detail="学生成绩不存在")
+    return {"msg":"删除成功"}
@@ -0,0 +1,50 @@
+from fastapi import APIRouter,Depends,HTTPException
+from sqlalchemy.orm import Session
+from sqlalchemy_fastapi_demo.database import get_db
+from sqlalchemy_fastapi_demo.scheme.advisors import AdvisorIn, AdvisorOut
+from sqlalchemy_fastapi_demo.dao.advisor import service,crud
+
+
+router = APIRouter()
+@router.get("/read",response_model=AdvisorOut)
+def read_advisor(advisor_id:int|None=None,advisor_name:str|None=None,db:Session=Depends(get_db)):
+    r_advisor=crud.search_advisor(db=db,advisor_id=advisor_id,advisor_name=advisor_name)
+    if r_advisor is not None:
+        return r_advisor
+    raise HTTPException(status_code=404,detail="顾问老师不存在")
+@router.get("/list",response_model=list[AdvisorOut])
+def list_advisor(advisor_name:str|None=None,db:Session=Depends(get_db)):
+    return crud.list_advisors(db=db,advisor_name=advisor_name)
+    # r_advisor=crud.list_advisors(db=db,advisor_name=advisor_name,phone=phone)
+    # if r_advisor:
+    #     return r_advisor
+    # raise HTTPException(status_code=404,detail="顾问老师不存在")
+@router.post("/create",response_model=AdvisorOut)
+def create_advisor(advisor_in:AdvisorIn,db:Session=Depends(get_db)):
+    c_advisor,error=service.create_advisor(db=db,advisor_id=advisor_in.advisor_id,advisor_name=advisor_in.advisor_name)
+    if error == "DUPLICATE":
+        raise HTTPException(status_code=409, detail="该顾问ID已被占用")
+    if error is not None:
+        # 兜底：Service 新增了返回码但这里没接住，不要静默当成成功
+        raise HTTPException(status_code=500, detail="未处理的业务结果: " + error)
+    return c_advisor
+
+
+@router.put("/update",response_model=AdvisorOut)
+def update_advisor(advisor_in:AdvisorIn,db:Session=Depends(get_db)):
+    u_advisor,error=service.update_advisor(db=db,advisor_id=advisor_in.advisor_id,advisor_in=advisor_in)
+    if error == "NOT_FOUND":
+        raise HTTPException(status_code=404,detail="更新失败：顾问不存在")
+    if error is not None:
+        raise HTTPException(status_code=500, detail="未处理的业务结果: " + error)
+    return u_advisor
+@router.delete("/delete/{advisor_id}")
+def delete_advisor(advisor_id:int,db:Session=Depends(get_db)):
+    d_advisor,error=service.delete_advisor(db=db,advisor_id=advisor_id)
+    if error == "NOT_FOUND":
+        raise HTTPException(status_code=404,detail="顾问老师不存在")
+    if error == "HAS_STUDENTS":
+        raise HTTPException(status_code=409, detail="顾问老师下还有学生")
+    if error is not None:
+        raise HTTPException(status_code=500, detail="未处理的业务结果: " + error)
+    return {"message":"删除成功","delete_id":d_advisor}
@@ -0,0 +1,74 @@
+# # api/c_lass.py
+# # 本文件定义班级的所有 API 路由（Controller 层）
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+#  APIRouter=创建路由，Depends=依赖注入，HTTPException=错误响应，Query=查询参数校验
+from sqlalchemy.orm import Session  #  会话类型
+from sqlalchemy_fastapi_demo.database import get_db  #  导入会话生成器
+from sqlalchemy_fastapi_demo.dao.c_lass import create_c_lass, get_c_lass, get_all_c_lass, update_c_lass, delete_c_lass
+#  把DAO层的5个函数都导入进来，接口里直接调用
+from sqlalchemy_fastapi_demo.scheme.c_lass import ClassCreate, ClassUpdate, ClassResponse
+
+# 路由对象，具体前缀（/c_lass）在 main.py 统一注册
+router = APIRouter()#  创建本模块的路由对象，main.py会把它注册进应用
+
+
+# ---------- 1. 新增班级 POST ----------
+# 完整地址：POST /c_lass/add
+@router.post("/add", response_model=ClassResponse, summary="新增班级")
+#  装饰器：注册一个POST接口，路径/add；response_model=返回时按ClassResponse格式化
+def add_c_lass(class_data: ClassCreate, db: Session = Depends(get_db)):
+    #  class_data: ClassCreate = FastAPI自动校验请求体并转成对象；db = FastAPI自动注入会话
+    #  先查重：如果这个班级ID已经存在（包括逻辑删除的），进入if
+    if get_c_lass(db, class_data.class_id):
+        raise HTTPException(status_code=400, detail="该班级ID已存在，不能重复添加")
+    return create_c_lass(db, class_data)#  通过查重，调DAO层真正插入，返回结果
+
+
+# ---------- 2. 根据id查询单个班级 GET ----------
+# 完整地址：GET /c_lass/{class_id}
+@router.get("/{class_id}", response_model=ClassResponse, summary="根据id查询班级")
+#  GET接口，路径里带班级编号，如 GET /c_lass/101
+def get_class_by_id(class_id: int, db: Session = Depends(get_db)):    #  调DAO查询
+    db_class = get_c_lass(db, class_id)
+    if not db_class:
+        raise HTTPException(status_code=404, detail="该班级不存在")    #  查不到就返回404，带提示信息
+    return db_class
+
+
+# ---------- 3. 分页查询所有班级 GET ----------
+# 完整地址：GET /c_lass/?skip=0&limit=10
+@router.get("/", response_model=list[ClassResponse], summary="分页查询班级列表")
+# GET根路径，如 GET /c_lass/?skip=0&limit=10；response_model是列表
+def get_class_list(
+    skip: int = Query(0, ge=0, description="跳过的记录数"),  #  查询参数skip，默认0，必须≥0
+    limit: int = Query(10, ge=1, le=200, description="每页最大记录数"),  #  查询参数limit，默认10，范围1~200
+    db: Session = Depends(get_db),
+):
+    return get_all_c_lass(db, skip, limit)  #  调DAO分页查询
+
+
+# ---------- 4. 修改班级 PUT ----------
+# 完整地址：PUT /c_lass/{class_id}
+@router.put("/{class_id}/", response_model=ClassResponse, summary="修改班级信息")
+def update_class(class_id: int, update_data: ClassUpdate, db: Session = Depends(get_db)):
+# def update_class(class_id: int, start_time: date, db: Session = Depends(get_db)):
+
+    if not get_c_lass(db, class_id): #  先确认要改的班级存在（存在且没删）
+        raise HTTPException(status_code=404, detail="要修改的班级不存在")
+    return update_c_lass(db, class_id, update_data) #  调DAO执行局部更新
+
+
+# ---------- 5. 删除班级 DELETE ----------
+# 完整地址：DELETE /c_lass/{class_id}
+@router.delete("/{class_id}", summary="删除班级")
+def delete_class(class_id: int, db: Session = Depends(get_db)):
+    if not delete_c_lass(db, class_id):
+        raise HTTPException(status_code=404, detail="该班级不存在")
+    #  删除失败（查不到）返回404
+    return {"message": "删除成功"}
+
+
+
+
+#API 层只做三件事—— 接请求、校验 / 查重、调 DAO，自己不写 SQL
@@ -0,0 +1,116 @@
+# api/employment_api.py
+# 本文件定义学生就业信息相关的所有 API 路由（Controller 层）
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+from typing import List, Optional
+
+from sqlalchemy_fastapi_demo.database import get_db
+from sqlalchemy_fastapi_demo.dao.employment import EmploymentBaseDAO,EmploymentOfferDAO
+from sqlalchemy_fastapi_demo.model.employment import EmploymentBase, EmploymentOffer
+from sqlalchemy_fastapi_demo.scheme.employment import EmploymentOfferUpdate, EmploymentBaseUpdate, EmploymentBaseCreate,EmploymentOfferCreate,EmploymentQuery,EmploymentOfferQueryResponse,EmploymentBaseQueryResponse
+router = APIRouter()
+
+#-----------多条件查询就业协议信息-----------
+@router.get("/offer/search/",response_model=List[EmploymentOfferQueryResponse],summary="多条件查询就业协议信息")
+def search_offer(
+    stu_id: Optional[int] = Query(None, description="学生学号"),
+    offer_id: Optional[int] = Query(None, description="offer编号，需同时提供学生编号"),
+    skip: int = Query(0, ge=0, description="跳过的记录条数"),
+    limit: int = Query(10, ge=1, le=200, description="返回最大记录数"),
+    db: Session = Depends(get_db)
+):
+    #捕获异常
+    try:
+        data_list = EmploymentOfferDAO.different_choice_query(
+            db=db, stu_id=stu_id, offer_id=offer_id, skip=skip, limit=limit
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    return data_list
+#-----------多条件查询就业基础信息-----------
+@router.get("/base/search/",response_model=List[EmploymentBaseQueryResponse],summary="多条件查询就业基础信息")
+def search_base(
+    query: EmploymentQuery = Depends(),
+    skip: int = Query(0, ge=0, description="跳过的记录条数"),
+    limit: int = Query(10, ge=1, le=200, description="返回最大记录数"),
+    db: Session = Depends(get_db)
+):
+    # 多条件筛选就业基础信息 + 分页
+    if query.min_salary is not None and query.max_salary is not None and query.min_salary > query.max_salary:
+        raise HTTPException(status_code=400, detail="最低薪资不能大于最高薪资")
+
+    data_list = EmploymentBaseDAO.different_choice_query(
+        db=db,
+        stu_id=query.stu_id,
+        company_name=query.company_name,
+        min_salary=query.min_salary,
+        max_salary=query.max_salary,
+        skip=skip,
+        limit=limit
+    )
+    return data_list
+#-----------添加就业基础信息-----------
+@router.post("/base/add/",response_model=EmploymentBaseQueryResponse,summary="添加就业基础信息")
+def add_base(obj: EmploymentBaseCreate, db: Session = Depends(get_db)):
+    items = EmploymentBaseDAO.get_by_stu_id(db, obj.stu_id)
+    if items:
+        raise HTTPException(status_code=400, detail="该学生的就业基础信息已存在，不可重复新增")
+
+    new_obj = EmploymentBase(
+        stu_id=obj.stu_id,
+        employment_open_time=obj.employment_open_time,
+        job_time=obj.job_time,
+        company_name=obj.company_name,
+        salary=obj.salary,
+        is_deleted=0
+    )
+    res = EmploymentBaseDAO.create(db, new_obj)
+    return res
+#-----------添加就业协议信息-----------
+@router.post("/offer/add/",response_model=EmploymentOfferQueryResponse,summary="添加就业协议信息")
+def add_offer(obj: EmploymentOfferCreate, db: Session = Depends(get_db)):
+    new_obj = EmploymentOffer(
+        stu_id=obj.stu_id,
+        offer_id=obj.offer_id,
+        offer_time=obj.offer_time,
+        is_deleted=0
+    )
+    res = EmploymentOfferDAO.create(db, new_obj)
+    return res
+#-----------修改就业基础信息-----------
+@router.put("/base/{stu_id}/",response_model=EmploymentBaseQueryResponse,summary="修改就业基础信息")
+def update_base(stu_id: int, obj: EmploymentBaseUpdate, db: Session = Depends(get_db)):
+    update_data = obj.model_dump(exclude_unset=True)
+    try:
+        res = EmploymentBaseDAO.update(db, stu_id, update_data)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    if not res:
+        raise HTTPException(status_code=404, detail="未找到该学生就业基础信息")
+    return res
+#-----------修改就业协议信息-----------
+@router.put("/offer/{offer_id}", response_model=EmploymentOfferQueryResponse,summary="修改就业协议信息")
+def update_offer(stu_id:int,offer_id: int, obj: EmploymentOfferUpdate, db: Session = Depends(get_db)):
+    # offer 表主键是 (stu_id, offer_id)，定位记录需要两者
+    try:
+        res = EmploymentOfferDAO.update(db, stu_id, offer_id, obj.offer_time)
+    except ValueError as e:
+        # 捕获校验异常，返回标准400错误给前端
+        raise HTTPException(status_code=400, detail=str(e))
+    if not res:
+        raise HTTPException(status_code=404, detail="未找到该就业协议记录")
+    return res
+#-----------逻辑删除就业基础信息-----------
+@router.delete("/base/{stu_id}/",summary="删除就业基础信息")
+def delete_base(stu_id: int, db: Session = Depends(get_db)):
+    ok = EmploymentBaseDAO.delete(db, stu_id)
+    if not ok:
+        raise HTTPException(status_code=404, detail="记录不存在或已经删除")
+    return {"code": 200, "msg": "删除成功"}
+#-----------逻辑删除就业协议信息-----------
+@router.delete("/offer/{stu_id}/{offer_id}",summary="修改就业协议信息")
+def delete_offer(stu_id: int, offer_id: int, db: Session = Depends(get_db)):
+    ok = EmploymentOfferDAO.delete(db, stu_id, offer_id)
+    if not ok:
+        raise HTTPException(status_code=404, detail="记录不存在或已经删除")
+    return {"code": 200, "msg": "删除成功"}
@@ -0,0 +1,105 @@
+from fastapi import APIRouter,Depends,HTTPException
+from sqlalchemy.orm import Session
+from sqlalchemy import func
+from sqlalchemy_fastapi_demo.scheme.statistics import StudentAge, ClassCount, ScoreCount, ScoreAvg, Employment, EmploymentOff
+from sqlalchemy_fastapi_demo.database import get_db
+router = APIRouter()
+from sqlalchemy_fastapi_demo.model.students import Student
+from sqlalchemy_fastapi_demo.model.Score import Score
+from sqlalchemy_fastapi_demo.model.employment import EmploymentBase,EmploymentOffer
+from sqlalchemy_fastapi_demo.model.c_lass import Classinfo
+from fastapi import FastAPI, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+
+from sqlalchemy_fastapi_demo.database import get_db
+from sqlalchemy_fastapi_demo.dao.statistics import (students_age, class_statistics,score_above, score_fail,
+avg_scores,top_salary,stu_every,class_avg)
+# app = FastAPI(title="统计接口")
+
+# -动态年龄范围查询**：支持用户输入年龄阈值及比较条件（如大于、小于、等于、区间等），动态查询符合条件的学员信息。
+@router.get("/students/age")
+def api_students_age(op: str = Query(..., description=">, <, =, >=, <=, between"),
+    age_value: int = Query(None),age_min: int = Query(None),
+    age_max: int = Query(None),db: Session = Depends(get_db)):
+    try:
+        rows = students_age(db, op, age_value, age_min, age_max)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    return [{"stu_id": s.stu_id,
+        "stu_name": s.stu_name,
+        "age": s.age,
+        "gender": s.gender,
+        "class_id": s.class_id,} for s in rows]
+
+# 多维度班级统计**：统计每个班级的总人数，以及按性别（男、女）细分的人数分布。
+@router.get("/classes/statistics")
+def api_class_statistics(db: Session = Depends(get_db)):
+    return class_statistics(db)
+
+
+# - 查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩。
+@router.get("/scores/above")
+def api_score_above(
+    score_input: int = Query(..., ge=0, le=100),
+    db: Session = Depends(get_db)):
+    rows = score_above(db, score_input)
+    return [{"stu_id": i[0], "stu_name": i[1], "exam_order": i[2], "score": i[3]} for i in rows]
+
+# - 查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细。
+@router.get("/scores/fail")
+def api_score_fail(
+    fail_times: int = Query(..., ge=1),
+    fail_score: int = Query(60, ge=0, le=100),
+    db: Session = Depends(get_db)):
+    rows = score_fail(db, fail_times, fail_score)
+    return [{"stu_name": i[0], "class_id": i[1], "exam_order": i[2], "score": i[3]} for i in rows]
+
+# 统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序
+@router.get("/scores/avg")
+def api_avg_scores(order: str = Query("desc", pattern="^(asc|desc)$"),
+    db: Session = Depends(get_db)):
+    a = avg_scores(db, order)
+    return [{"exam_id": i.exam_id,
+             "class_id": i.class_id,
+            "avg_score": round(float(i.avg_score), 2)} for i in a]
+
+#就业统计
+# - 统计就业薪资排名 Top N（动态输入 N）的学生的姓名、班级、就业时间和就业公司。
+@router.get("/employment/top")
+def api_top_salary(n: int = Query(5, ge=1), db: Session = Depends(get_db)):
+    return top_salary(db, n)
+
+# - 统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间）。
+@router.get("/employment/every")
+def api_stu_every(db: Session = Depends(get_db)):
+    return stu_every(db)
+
+# 每个班级的平均就业时长
+@router.get("/employment/class_avg")
+def api_class_duration(db: Session = Depends(get_db)):
+    return class_avg(db)
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+