上传文件至「api」
This commit is contained in:
@@ -0,0 +1,78 @@
|
||||
from fastapi import APIRouter,Depends,HTTPException,Query
|
||||
from dao.class_dao import add_class_dao, update_class_dao, delete_class_dao, hard_delete_class_dao,class_no_exist,get_class_list_dao,get_class_by_id_dao
|
||||
from schema.class_schema import ClassCreate,ClassResponse,ClassUpdate
|
||||
from database import get_db
|
||||
from typing import Optional
|
||||
from sqlalchemy.orm import Session
|
||||
class_router = APIRouter()
|
||||
@class_router.post('',response_model=ClassResponse,summary='添加班级信息')
|
||||
def add_class(c:ClassCreate,db=Depends(get_db)):
|
||||
if class_no_exist(c.class_id,db):
|
||||
raise HTTPException(status_code=404,detail='该编号已经存在')
|
||||
ok,result = add_class_dao(c.model_dump(),db)
|
||||
if not ok:
|
||||
raise HTTPException(status_code=500,detail='服务器繁忙或指定的老师不存在')
|
||||
return ClassResponse(totals=1,data=result)
|
||||
@class_router.put('/{id}',summary='更新班级信息')
|
||||
def update_class(id:int,c:ClassUpdate,db=Depends(get_db)):
|
||||
d = c.model_dump()
|
||||
r = update_class_dao(id=id ,update_data=d,db=db)
|
||||
if not r:
|
||||
raise HTTPException(status_code=500,detail='更新失败')
|
||||
return{'code':200,'total':1,'detail':'更新成功'}
|
||||
'''
|
||||
@class_app.get('')
|
||||
def get_class(id:int,db=Depends(get_db),page:int=1,page_size:int=10):
|
||||
rows = get_class_dao(id=id,db=db,page=page,page_size=page_size)
|
||||
if rows:
|
||||
return rows
|
||||
raise HTTPException(status_code=404,detail='该班级不存在')
|
||||
'''
|
||||
@class_router.get('',summary='查询班级信息')
|
||||
def get_classes(
|
||||
id: Optional[int] = Query(None, description="班级ID"),
|
||||
class_id: Optional[str] = Query(None, description="班级编号"),
|
||||
class_name: Optional[str] = Query(None, description="班级名称（支持模糊搜索）"),
|
||||
ht_id: Optional[int] = Query(None, description="班主任ID"),
|
||||
t_id: Optional[int] = Query(None, description="教师ID"),
|
||||
page: int = Query(1, ge=1, description="页码"),
|
||||
page_size: int = Query(10, ge=1, le=100, description="每页条数"),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
# 如果明确传了 id，当作单条查询处理
|
||||
if id is not None:
|
||||
single_class = get_class_by_id_dao(id=id, db=db)
|
||||
if not single_class:
|
||||
raise HTTPException(status_code=404, detail="该班级不存在")
|
||||
return single_class
|
||||
|
||||
# 多条件列表查询
|
||||
total, items = get_class_list_dao(
|
||||
db=db,
|
||||
class_id=class_id,
|
||||
class_name=class_name,
|
||||
ht_id=ht_id,
|
||||
t_id=t_id,
|
||||
page=page,
|
||||
page_size=page_size
|
||||
)
|
||||
|
||||
return {
|
||||
"total": total,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
"items": items
|
||||
}
|
||||
@class_router.delete('/{id}',summary='软删除') #软删除
|
||||
def delete_class(id:int,db=Depends(get_db)):
|
||||
rows = delete_class_dao(id=id,db=db)
|
||||
if not rows:
|
||||
raise HTTPException(status_code=404,detail='该班级不存在或已删除')
|
||||
return{'code':200,'total':rows,'detail':'删除成功'}
|
||||
|
||||
@class_router.delete('{id}/hard',summary='删除班级信息')
|
||||
def hard_delete_class(id:int,db=Depends(get_db)):
|
||||
rows = hard_delete_class_dao(id=id,db=db)
|
||||
if not rows:
|
||||
raise HTTPException(status_code=404,detail='该班级不存在')
|
||||
return{'code':200,'total':rows,'detail':'物理删除成功'}
|
||||
@@ -0,0 +1,35 @@
|
||||
from fastapi import APIRouter,Depends,HTTPException
|
||||
from database import *
|
||||
from schema.epy_schema import EmploymentRequest
|
||||
from dao.epy_dao import get_employment_dao,wq_employment_dao,upd_employment_dao,del_employment_dao
|
||||
epy_router = APIRouter()
|
||||
|
||||
|
||||
@epy_router.get('',summary='查询就业信息')
|
||||
def get_employment(stu_id:str,company:str,salary:int,db=Depends(get_db)): # get变动
|
||||
try:
|
||||
rows = get_employment_dao(db, stu_id, company, salary)
|
||||
if not rows: # dao层返回的是q.() 是列表
|
||||
return {'code':404,'detail':'数据不存在','total':[]} # 数据不存在返回报错信息
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f'查询失败:{str(e)}')
|
||||
return {'code':200,'detail':'查询成功','date':[{'stu_id':i.stu_id,'company':i.company,'salary':i.salary} for i in rows]}
|
||||
|
||||
|
||||
@epy_router.post('',summary='新增就业信息')
|
||||
def add_employment(e:EmploymentRequest,db=Depends(get_db)):
|
||||
d= e.model_dump()
|
||||
r = wq_employment_dao(d,db)
|
||||
return {'code':200,'detail':'添加成功!','total':r}
|
||||
#
|
||||
#
|
||||
@epy_router.put('',summary='更新就业信息')
|
||||
def update_employment(e:EmploymentRequest,db=Depends(get_db)):
|
||||
e1 = e.model_dump()
|
||||
r = upd_employment_dao(e1,db)
|
||||
return {'code':200,'msg':'更新成功','total':r}
|
||||
#
|
||||
@epy_router.delete('',summary='删除就业信息')
|
||||
def delete_employment(stu_id:str,db=Depends(get_db)):
|
||||
r =del_employment_dao(stu_id,db)
|
||||
return {'code':200,'totals':r,'detail':'删除成功!'}
|
||||
@@ -0,0 +1,45 @@
|
||||
from database import get_db
|
||||
from fastapi import APIRouter,Depends
|
||||
from dao.sc_dao import a,a1,a2,a3,a4,a5
|
||||
from schema.sc_schema import Test
|
||||
|
||||
score_router=APIRouter()
|
||||
@score_router.post('',summary='添加成绩信息')
|
||||
def scores(s:Test,db=Depends(get_db)):
|
||||
d = s.model_dump()
|
||||
a(d,db)
|
||||
return {'code':200,'detail':'添加成功！',}
|
||||
@score_router.delete('',summary='删除成绩信息')
|
||||
def scores1(stu_id:int,db=Depends(get_db)):
|
||||
r = a1(stu_id,db)
|
||||
if r==0:
|
||||
return '没有该学生'
|
||||
else:
|
||||
return f'已删除学生编号为{stu_id}的学生'
|
||||
@score_router.put('',summary='更新成绩信息')
|
||||
def scores2(stu_id:int,exam_seq:int,s:Test,db=Depends(get_db)):
|
||||
r=a2(stu_id,exam_seq,s,db)
|
||||
if r!=0:
|
||||
return '更新成功'
|
||||
else:
|
||||
return '没有该数据'
|
||||
@score_router.get('',summary='通过学号查询成绩信息')
|
||||
def scores3(stu_id:int,exam_seq:int,db=Depends(get_db)):
|
||||
r=a3(stu_id,exam_seq,db)
|
||||
try:
|
||||
return {'成绩编号':r.id,'学生编号':r.stu_id,f'第{exam_seq}次成绩:':r.score,'创建日期':r.create_date,'更新日期':r.update_date}
|
||||
except:
|
||||
return '没有该学生'
|
||||
@score_router.delete('/{stu_id}',summary='软删除')
|
||||
def scores4(stu_id:int,exam_seq:int,db=Depends(get_db)):
|
||||
r=a4(stu_id, exam_seq, db)
|
||||
if r == 1:
|
||||
return '没有该数据'
|
||||
else:
|
||||
return '已删除'
|
||||
@score_router.get('/{stu_id}',summary='计算总分、平均分、最大值、最小值')
|
||||
def scores5(stu_id:int,db=Depends(get_db)):
|
||||
s,r,a=a5(stu_id,db)
|
||||
|
||||
return f'总分:{s},平均分：{s/len(r)},最大值：{max(a)},最小值：{min(a)}'
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
from database import get_db
|
||||
from dao.stu_dao import create_student, get_student_by_id, get_student_list, update_student, delete_student_logic
|
||||
from schema.stu_schema import (
|
||||
StudentCreateRequest,
|
||||
StudentUpdateRequest,
|
||||
StudentQuery,
|
||||
StudentResponse,
|
||||
StudentPageResponse
|
||||
)
|
||||
|
||||
# 创建路由对象 API Router1
|
||||
stu_router = APIRouter()
|
||||
@stu_router.post("", response_model=StudentResponse, summary="新增学生")
|
||||
def add_student(
|
||||
student_req: StudentCreateRequest,
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
db_stu = create_student(db, student_req)
|
||||
if db_stu is None:
|
||||
raise HTTPException(status_code=500, detail="新增学生失败，数据库异常")
|
||||
return StudentResponse.model_validate(db_stu)
|
||||
|
||||
|
||||
@stu_router.get("", response_model=StudentPageResponse, summary="根据学号姓名班级查询")
|
||||
def get_student_page(
|
||||
query: StudentQuery = Depends(),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
total, db_stu_list = get_student_list(
|
||||
db,
|
||||
stu_id=query.stu_id,
|
||||
stu_name=query.stu_name,
|
||||
class_id=query.class_id,
|
||||
page=query.page,
|
||||
page_size=query.page_size
|
||||
)
|
||||
if total == 0:
|
||||
raise HTTPException(status_code=404, detail="没有找到符合条件的学生")
|
||||
|
||||
item_list = [StudentResponse.model_validate(s) for s in db_stu_list]
|
||||
|
||||
return StudentPageResponse(
|
||||
total=total,
|
||||
page=query.page,
|
||||
page_size=query.page_size,
|
||||
items=item_list
|
||||
)
|
||||
|
||||
|
||||
@stu_router.get("/{stu_id}", response_model=StudentResponse, summary="根据学号查询")
|
||||
def get_one_student(
|
||||
stu_id: str,
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
db_stu = get_student_by_id(db, stu_id)
|
||||
if db_stu is None:
|
||||
raise HTTPException(status_code=404, detail="该学生不存在")
|
||||
return StudentResponse.model_validate(db_stu)
|
||||
|
||||
@stu_router.put("/{stu_id}", response_model=StudentResponse, summary="修改学生信息")
|
||||
def edit_student(
|
||||
stu_id: str,
|
||||
update_req: StudentUpdateRequest,
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
exist = get_student_by_id(db, stu_id)
|
||||
if exist is None:
|
||||
raise HTTPException(status_code=404, detail="该学生不存在")
|
||||
|
||||
db_stu = update_student(db, stu_id, update_req)
|
||||
if db_stu is None:
|
||||
raise HTTPException(status_code=500, detail="修改学生失败，数据库异常")
|
||||
|
||||
return StudentResponse.model_validate(db_stu)
|
||||
|
||||
|
||||
@stu_router.delete("/{stu_id}", response_model=StudentResponse, summary="删除学生信息")
|
||||
def remove_student(
|
||||
stu_id: str,
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
exist = get_student_by_id(db, stu_id)
|
||||
if exist is None:
|
||||
raise HTTPException(status_code=404, detail="该学生不存在")
|
||||
|
||||
db_stu = delete_student_logic(db, stu_id)
|
||||
if db_stu is None:
|
||||
raise HTTPException(status_code=500, detail="删除学生失败，数据库异常")
|
||||
|
||||
return StudentResponse.model_validate(db_stu)
|
||||
@@ -0,0 +1,59 @@
|
||||
from fastapi import APIRouter,Depends,UploadFile,File,Query
|
||||
from dao.th_dao import insert_teachers_dao,update_teachers_dao,delete_teachers_dao,get_teachers_limit_dao,get_teachers_name_dao
|
||||
from schema.th_schema import RequestModel, ResponseModel
|
||||
from database import get_db
|
||||
|
||||
t_router = APIRouter()
|
||||
|
||||
@t_router.get("",summary='分页查询教师接口')
|
||||
def get_teachers(p:int=Query(default=0,ge=0,description='页码,0表示查询全部'),
|
||||
n:int=Query(default=0,ge=0,description='每页条数,0表示查询全部'),
|
||||
db=Depends(get_db)):
|
||||
db_teachers = get_teachers_limit_dao(p,n,db)
|
||||
return ResponseModel(data=db_teachers,total=len(db_teachers))
|
||||
|
||||
@t_router.get("/{name}",summary='通过名称查询教师接口')
|
||||
def get_name_teachers(name:str,db=Depends(get_db)):
|
||||
db_teachers = get_teachers_name_dao(name,db)
|
||||
return ResponseModel(data=db_teachers,total=len(db_teachers))
|
||||
|
||||
@t_router.post("",summary='新增教师接口')
|
||||
def insert_teachers(req:RequestModel,db=Depends(get_db)):
|
||||
rq= req.model_dump()
|
||||
n,m = insert_teachers_dao(rq,db)
|
||||
# print('打印获取的值:',n,m)
|
||||
return ResponseModel(total=n,msg=m)
|
||||
|
||||
@t_router.put("",summary='更新教师接口')
|
||||
def update_teachers(req:RequestModel,db=Depends(get_db)):
|
||||
rq= req.model_dump()
|
||||
n,m = update_teachers_dao(rq,db)
|
||||
return ResponseModel(total=n,msg=m)
|
||||
|
||||
@t_router.delete("/{t_id}",summary='删除教师接口')
|
||||
def delete_teachers(t_id:int,db=Depends(get_db)):
|
||||
n,m =delete_teachers_dao(t_id,db)
|
||||
return ResponseModel(msg=m,total=n)
|
||||
|
||||
@t_router.post('/files',summary='上传文件接口')
|
||||
async def upload_files(f:UploadFile=File(),db=Depends(get_db)):
|
||||
# print('UploadFile上传文件,读取该对象:',f)
|
||||
text =await f.read()
|
||||
l =text.decode('utf-8').split('\n')
|
||||
lg = 0
|
||||
dt = []
|
||||
print('长度为',lg,l)
|
||||
for i in l:
|
||||
if '教师编号' in i:
|
||||
continue
|
||||
p = i.split()
|
||||
if not p: # 判断如果p为[],跳过
|
||||
continue
|
||||
l2 = ['t_id','t_name','t_age','t_phone','t_sex','class_id']
|
||||
d =dict()
|
||||
for j in range(len(p)):
|
||||
d[l2[j]]=p[j]
|
||||
insert_teachers_dao(d,db)
|
||||
lg = lg + 1
|
||||
dt.append(d)
|
||||
return ResponseModel(msg='上传成功',total=lg,data=dt)
|
||||
Reference in New Issue
Block a user